From 54c41cb2c3453fa17223c9d4ee264e6f22c654ce Mon Sep 17 00:00:00 2001
From: BradleyOlson64 <lotrftw9@gmail.com>
Date: Fri, 27 Oct 2023 12:29:20 -0700
Subject: [PATCH 01/10] assigner_bulk compile fixes

---
 Cargo.lock                                    |   1 +
 .../src/runtime/scheduler.md                  |   1 +
 polkadot/runtime/parachains/Cargo.toml        |   1 +
 .../parachains/src/assigner_bulk/mod.rs       | 102 +++++++++---------
 4 files changed, 57 insertions(+), 48 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index e0ca0b012c64..36f8f7fcd3e3 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -12534,6 +12534,7 @@ dependencies = [
  "pallet-authorship",
  "pallet-babe",
  "pallet-balances",
+ "pallet-broker",
  "pallet-message-queue",
  "pallet-session",
  "pallet-staking",
diff --git a/polkadot/roadmap/implementers-guide/src/runtime/scheduler.md b/polkadot/roadmap/implementers-guide/src/runtime/scheduler.md
index 2b8832946fc9..55d4db1a7ca6 100644
--- a/polkadot/roadmap/implementers-guide/src/runtime/scheduler.md
+++ b/polkadot/roadmap/implementers-guide/src/runtime/scheduler.md
@@ -182,6 +182,7 @@ struct CoreAssignment {
   core: CoreIndex,
   para_id: ParaId,
   kind: AssignmentKind,
+  group_idx: GroupIndex,
 }
 // reasons a core might be freed.
 enum FreedReason {
diff --git a/polkadot/runtime/parachains/Cargo.toml b/polkadot/runtime/parachains/Cargo.toml
index 9f1ec57257f8..f24b479413a4 100644
--- a/polkadot/runtime/parachains/Cargo.toml
+++ b/polkadot/runtime/parachains/Cargo.toml
@@ -32,6 +32,7 @@ pallet-authority-discovery = { path = "../../../substrate/frame/authority-discov
 pallet-authorship = { path = "../../../substrate/frame/authorship", default-features = false }
 pallet-balances = { path = "../../../substrate/frame/balances", default-features = false }
 pallet-babe = { path = "../../../substrate/frame/babe", default-features = false }
+pallet-broker = { path = "../../../substrate/frame/broker", default-features = false }
 pallet-message-queue = { path = "../../../substrate/frame/message-queue", default-features = false }
 pallet-session = { path = "../../../substrate/frame/session", default-features = false }
 pallet-staking = { path = "../../../substrate/frame/staking", default-features = false }
diff --git a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
index b78f11a1f3a2..1a2ddcb40c06 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
@@ -18,16 +18,10 @@
 //!
 //! Handles scheduling of bulk core time.
 
-mod benchmarking;
-mod mock_helpers;
-
-#[cfg(test)]
-mod tests;
-
 use crate::{
 	assigner_on_demand, configuration, paras,
 	scheduler::common::{
-		Assignment, AssignmentProvider, AssignmentProviderConfig, AssignmentVersion, V0Assignment,
+		Assignment, AssignmentProvider, AssignmentProviderConfig, AssignmentVersion,
 	},
 };
 
@@ -45,6 +39,7 @@ use sp_runtime::{
 	traits::{One, SaturatedConversion},
 	FixedPointNumber, FixedPointOperand, FixedU128, Perbill, Saturating,
 };
+use pallet_broker::CoreAssignment;
 
 use sp_std::{collections::vec_deque::VecDeque, prelude::*};
 
@@ -65,6 +60,7 @@ impl WeightInfo for TestWeightInfo {}
 /// AssignmentSets as they are scheduled by block number
 ///
 /// for a particular core.
+#[derive(Encode, Decode, TypeInfo)]
 struct Schedule<N> {
 	// Original assignments
 	assignments: Vec<(CoreAssignment, PartsOf57600)>,
@@ -83,6 +79,7 @@ struct Schedule<N> {
 ///
 /// This is the state of assignments currently being served via the `AssignmentProvider` interface,
 /// as opposed to `Schedule` which is upcoming not yet served assignments.
+#[derive(Encode, Decode, TypeInfo)]
 struct CoreState<N> {
 	/// Assignments with current state.
 	///
@@ -109,6 +106,7 @@ struct CoreState<N> {
 	step: PartsOf57600,
 }
 
+#[derive(Encode, Decode, TypeInfo)]
 struct AssignmentState {
 	/// Ratio of the core this assignment has.
 	///
@@ -125,8 +123,8 @@ struct AssignmentState {
 impl<N> From<Schedule<N>> for CoreState<N> {
 	fn from(schedule: Schedule<N>) -> Self {
 		let Schedule { assignments, end_hint } = schedule;
-		let step = if let Some(step) = assignments.iter().min_by(|a, b| a.1.cmp(b.1)) {
-			step
+		let step = if let Some(min_step_assignment) = assignments.iter().min_by(|a, b| a.1.cmp(&b.1)) {
+			min_step_assignment.1
 		} else {
 			// Assignments empty, should not exist. In any case step size does not matter here:
 			log::debug!("assignments of a `Schedule` should never be empty.");
@@ -151,7 +149,7 @@ pub mod pallet {
 	pub struct Pallet<T>(_);
 
 	#[pallet::config]
-	pub trait Config: frame_system::Config + configuration::Config + paras::Config {
+	pub trait Config: frame_system::Config + configuration::Config + paras::Config + assigner_on_demand::Config {
 		/// Something that provides the weight of this pallet.
 		type WeightInfo: WeightInfo;
 	}
@@ -178,7 +176,7 @@ pub mod pallet {
 	/// Used for updating `end_hint` of that latest schedule based on newly appended schedules.
 	#[pallet::storage]
 	pub(super) type LatestCoreSchedule<T: Config> =
-		StorageMap<_, Twox256, CoreIndex, BlockNumberFor<N>, OptionQuery>;
+		StorageMap<_, Twox256, CoreIndex, BlockNumberFor<T>, OptionQuery>;
 
 	/// Assignments which are currently active.
 	///
@@ -190,7 +188,9 @@ pub mod pallet {
 
 	#[pallet::hooks]
 	impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
-		fn on_initialize(_now: BlockNumberFor<T>) -> Weight {}
+		fn on_initialize(_now: BlockNumberFor<T>) -> Weight {
+			unimplemented!("TODO: Implement")
+		}
 	}
 
 	#[pallet::call]
@@ -198,9 +198,10 @@ pub mod pallet {
 }
 
 /// Assignments as provided by our `AssignmentProvider` implementation.
-enum BulkAssignment<OnDemand> {
+#[derive(Encode, Decode, TypeInfo, Debug)]
+pub enum BulkAssignment<OnDemand> {
 	/// Assignment was an instantaneous core time assignment.
-	Instantenous(OnDemand),
+	Instantaneous(OnDemand),
 	/// Assignment was served directly from a core managed directly by bulk.
 	Bulk(ParaId),
 }
@@ -209,19 +210,19 @@ type BulkAssignmentType<T> = BulkAssignment<
 	<assigner_on_demand::Pallet<T> as AssignmentProvider<BlockNumberFor<T>>>::AssignmentType,
 >;
 
-impl Assignment for BulkAssignment {
+impl<OnDemand: Assignment> Assignment for BulkAssignment<OnDemand> {
 	fn para_id(&self) -> ParaId {
-		match &self {
-			Self::Instantenous(on_demand) => on_demand.para_id(),
-			Self::Bulk(para_id) => para_id,
+		match self {
+			Self::Instantaneous(on_demand) => on_demand.para_id(),
+			Self::Bulk(para_id) => *para_id,
 		}
 	}
 }
 
 impl<T: Config> AssignmentProvider<BlockNumberFor<T>> for Pallet<T> {
-	type AssignmentType = BulkAssignment;
+	type AssignmentType = BulkAssignmentType<T>;
 
-	type OldAssignmentType = BulkAssignment;
+	type OldAssignmentType = BulkAssignmentType<T>;
 
 	const ASSIGNMENT_STORAGE_VERSION: AssignmentVersion = AssignmentVersion::new(0);
 
@@ -241,37 +242,42 @@ impl<T: Config> AssignmentProvider<BlockNumberFor<T>> for Pallet<T> {
 
 		Workload::<T>::mutate(core_idx, |core_state| {
 			Self::ensure_workload(now, core_idx, core_state);
-
-			core_state.pos = core_state.pos % core_state.assignments.len();
-			let (a_type, a_state) = &mut core_state
-				.assignments
-				.get(core_state.pos)
-				.expect("We limited pos to the size of the vec one line above. qed");
-
-			// advance:
-			a_state.remaining -= core_state.step;
-			if a_state.remaining < step {
-				// Assignment exhausted, need to move to the next and credit remaining for next
-				// round.
-				core_state.pos += 1;
-				// Reset to ratio + still remaining "credits":
-				a_state.remaining += a_state.ratio;
-			}
-
-			match a_type {
-				CoreAssignment::Idle => return None,
-				CoreAssignment::Pool =>
-					return <assigner_on_demand::Pallet<T> as AssignmentProvider<
-						BlockNumberFor<T>,
-					>>::pop_assignment_for_core(core_idx),
-				CoreAssignment::Task(para_id) => return Some(BulkAssignment::Bulk(para_id)),
+			match core_state {
+				Some(core_state) => {
+					core_state.pos = core_state.pos % core_state.assignments.len() as u16;
+					let (a_type, a_state) = &mut core_state
+						.assignments
+						.get_mut(core_state.pos as usize)
+						.expect("We limited pos to the size of the vec one line above. qed");
+
+					// advance:
+					a_state.remaining -= core_state.step;
+					if a_state.remaining < core_state.step {
+						// Assignment exhausted, need to move to the next and credit remaining for next
+						// round.
+						core_state.pos += 1;
+						// Reset to ratio + still remaining "credits":
+						a_state.remaining += a_state.ratio;
+					}
+
+					match a_type {
+						CoreAssignment::Idle => return None,
+						CoreAssignment::Pool =>
+							return <assigner_on_demand::Pallet<T> as AssignmentProvider<
+								BlockNumberFor<T>,
+							>>::pop_assignment_for_core(core_idx)
+							.map(|assignment| BulkAssignment::Instantaneous(assignment)),
+						CoreAssignment::Task(para_id) => return Some(BulkAssignment::Bulk((*para_id).into())),
+					}
+				},
+				None => return None,
 			}
-		});
+		})
 	}
 
 	fn report_processed(assignment: Self::AssignmentType) {
 		match assignment {
-			BulkAssignment::Instantanous(on_demand) => <assigner_on_demand::Pallet<T> as AssignmentProvider<BlockNumberFor<T>>>::report_processed(on_demand),
+			BulkAssignment::Instantaneous(on_demand) => <assigner_on_demand::Pallet<T> as AssignmentProvider<BlockNumberFor<T>>>::report_processed(on_demand),
 			BulkAssignment::Bulk(_) => {}
 		}
 	}
@@ -312,7 +318,7 @@ impl<T: Config> Pallet<T> {
 			match workload.end_hint {
 				Some(end_hint) if end_hint < now => {
 					// Invariant: Always points to next item in `Workplan`, if such an item exists.
-					let n = end_hint.saturating_add(1);
+					let n = end_hint.saturating_add(BlockNumberFor::<T>::from(1u32));
 					// Workload expired - update to whatever is scheduled or `None` if nothing is:
 					Workplan::<T>::take((n, core_idx))
 				},
@@ -328,7 +334,7 @@ impl<T: Config> Pallet<T> {
 		};
 
 		// Needs update:
-		workload = update.map(|schedule| schedule.into());
+		*workload = update.map(|schedule| schedule.into());
 	}
 }
 

From ad161f0b9db8ffc1081c16f437e75156c4a7ef01 Mon Sep 17 00:00:00 2001
From: BradleyOlson64 <lotrftw9@gmail.com>
Date: Fri, 27 Oct 2023 14:42:40 -0700
Subject: [PATCH 02/10] Create test files

---
 .../src/assigner_bulk/mock_helpers.rs         | 86 ++++++++++++++++++
 .../parachains/src/assigner_bulk/mod.rs       |  4 +
 .../parachains/src/assigner_bulk/tests.rs     | 89 +++++++++++++++++++
 3 files changed, 179 insertions(+)
 create mode 100644 polkadot/runtime/parachains/src/assigner_bulk/mock_helpers.rs
 create mode 100644 polkadot/runtime/parachains/src/assigner_bulk/tests.rs

diff --git a/polkadot/runtime/parachains/src/assigner_bulk/mock_helpers.rs b/polkadot/runtime/parachains/src/assigner_bulk/mock_helpers.rs
new file mode 100644
index 000000000000..acfb24cbf194
--- /dev/null
+++ b/polkadot/runtime/parachains/src/assigner_bulk/mock_helpers.rs
@@ -0,0 +1,86 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot 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.
+
+// Polkadot 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 Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Helper functions for tests, also used in runtime-benchmarks.
+
+#![cfg(test)]
+
+use super::*;
+
+use crate::{
+	mock::MockGenesisConfig,
+	paras::{ParaGenesisArgs, ParaKind},
+};
+
+use primitives::{Balance, HeadData, ValidationCode};
+
+pub fn default_genesis_config() -> MockGenesisConfig {
+	MockGenesisConfig {
+		configuration: crate::configuration::GenesisConfig {
+			config: crate::configuration::HostConfiguration { ..Default::default() },
+		},
+		..Default::default()
+	}
+}
+
+#[derive(Debug)]
+pub struct GenesisConfigBuilder {
+	pub on_demand_cores: u32,
+	pub on_demand_base_fee: Balance,
+	pub on_demand_fee_variability: Perbill,
+	pub on_demand_max_queue_size: u32,
+	pub on_demand_target_queue_utilization: Perbill,
+	pub onboarded_on_demand_chains: Vec<ParaId>,
+}
+
+impl Default for GenesisConfigBuilder {
+	fn default() -> Self {
+		Self {
+			on_demand_cores: 10,
+			on_demand_base_fee: 10_000,
+			on_demand_fee_variability: Perbill::from_percent(1),
+			on_demand_max_queue_size: 100,
+			on_demand_target_queue_utilization: Perbill::from_percent(25),
+			onboarded_on_demand_chains: vec![],
+		}
+	}
+}
+
+impl GenesisConfigBuilder {
+	pub(super) fn build(self) -> MockGenesisConfig {
+		let mut genesis = default_genesis_config();
+		let config = &mut genesis.configuration.config;
+		config.on_demand_cores = self.on_demand_cores;
+		config.on_demand_base_fee = self.on_demand_base_fee;
+		config.on_demand_fee_variability = self.on_demand_fee_variability;
+		config.on_demand_queue_max_size = self.on_demand_max_queue_size;
+		config.on_demand_target_queue_utilization = self.on_demand_target_queue_utilization;
+
+		let paras = &mut genesis.paras.paras;
+		for para_id in self.onboarded_on_demand_chains {
+			paras.push((
+				para_id,
+				ParaGenesisArgs {
+					genesis_head: HeadData::from(vec![0u8]),
+					validation_code: ValidationCode::from(vec![0u8]),
+					para_kind: ParaKind::Parathread,
+				},
+			))
+		}
+
+		genesis
+	}
+}
diff --git a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
index 1a2ddcb40c06..7629daae9218 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
@@ -18,6 +18,10 @@
 //!
 //! Handles scheduling of bulk core time.
 
+mod mock_helpers;
+#[cfg(test)]
+mod tests;
+
 use crate::{
 	assigner_on_demand, configuration, paras,
 	scheduler::common::{
diff --git a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
new file mode 100644
index 000000000000..b7d0da3a2073
--- /dev/null
+++ b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
@@ -0,0 +1,89 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot 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.
+
+// Polkadot 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 Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+use super::*;
+
+use crate::{
+	assigner_bulk::mock_helpers::GenesisConfigBuilder,
+	initializer::SessionChangeNotification,
+	mock::{
+		new_test_ext, Balances, OnDemandAssigner, Paras, ParasShared, RuntimeOrigin, Scheduler,
+		System, Test,
+	},
+	paras::{ParaGenesisArgs, ParaKind},
+};
+use frame_support::{assert_noop, assert_ok, error::BadOrigin};
+use pallet_balances::Error as BalancesError;
+use primitives::{BlockNumber, SessionIndex, ValidationCode};
+use sp_std::collections::btree_map::BTreeMap;
+
+fn schedule_blank_para(id: ParaId, parakind: ParaKind) {
+	let validation_code: ValidationCode = vec![1, 2, 3].into();
+	assert_ok!(Paras::schedule_para_initialize(
+		id,
+		ParaGenesisArgs {
+			genesis_head: Vec::new().into(),
+			validation_code: validation_code.clone(),
+			para_kind: parakind,
+		}
+	));
+
+	assert_ok!(Paras::add_trusted_validation_code(RuntimeOrigin::root(), validation_code));
+}
+
+fn run_to_block(
+	to: BlockNumber,
+	new_session: impl Fn(BlockNumber) -> Option<SessionChangeNotification<BlockNumber>>,
+) {
+	while System::block_number() < to {
+		let b = System::block_number();
+
+		Scheduler::initializer_finalize();
+		Paras::initializer_finalize(b);
+
+		if let Some(notification) = new_session(b + 1) {
+			let mut notification_with_session_index = notification;
+			// We will make every session change trigger an action queue. Normally this may require
+			// 2 or more session changes.
+			if notification_with_session_index.session_index == SessionIndex::default() {
+				notification_with_session_index.session_index = ParasShared::scheduled_session();
+			}
+			Paras::initializer_on_new_session(&notification_with_session_index);
+			Scheduler::initializer_on_new_session(&notification_with_session_index);
+		}
+
+		System::on_finalize(b);
+
+		System::on_initialize(b + 1);
+		System::set_block_number(b + 1);
+
+		Paras::initializer_initialize(b + 1);
+		Scheduler::initializer_initialize(b + 1);
+
+		// In the real runtime this is expected to be called by the `InclusionInherent` pallet.
+		Scheduler::free_cores_and_fill_claimqueue(BTreeMap::new(), b + 1);
+	}
+}
+
+#[test]
+fn place_order_works() {
+	let alice = 1u64;
+	let para_id = ParaId::from(111);
+
+	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+		
+	});
+}

From f58615cf5b0ab24ec18dd01c9145171d37c73a11 Mon Sep 17 00:00:00 2001
From: BradleyOlson64 <lotrftw9@gmail.com>
Date: Fri, 27 Oct 2023 15:16:34 -0700
Subject: [PATCH 03/10] fmt

---
 .../parachains/src/assigner_bulk/mod.rs       | 28 +++++++++++--------
 .../parachains/src/assigner_parachains.rs     |  2 +-
 .../src/assigner_parachains/tests.rs          | 10 +++++--
 3 files changed, 24 insertions(+), 16 deletions(-)

diff --git a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
index 1a2ddcb40c06..b16d54e457c9 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
@@ -34,12 +34,12 @@ use frame_support::{
 	},
 };
 use frame_system::pallet_prelude::*;
+use pallet_broker::CoreAssignment;
 use primitives::{CoreIndex, Id as ParaId};
 use sp_runtime::{
 	traits::{One, SaturatedConversion},
 	FixedPointNumber, FixedPointOperand, FixedU128, Perbill, Saturating,
 };
-use pallet_broker::CoreAssignment;
 
 use sp_std::{collections::vec_deque::VecDeque, prelude::*};
 
@@ -123,13 +123,14 @@ struct AssignmentState {
 impl<N> From<Schedule<N>> for CoreState<N> {
 	fn from(schedule: Schedule<N>) -> Self {
 		let Schedule { assignments, end_hint } = schedule;
-		let step = if let Some(min_step_assignment) = assignments.iter().min_by(|a, b| a.1.cmp(&b.1)) {
-			min_step_assignment.1
-		} else {
-			// Assignments empty, should not exist. In any case step size does not matter here:
-			log::debug!("assignments of a `Schedule` should never be empty.");
-			1
-		};
+		let step =
+			if let Some(min_step_assignment) = assignments.iter().min_by(|a, b| a.1.cmp(&b.1)) {
+				min_step_assignment.1
+			} else {
+				// Assignments empty, should not exist. In any case step size does not matter here:
+				log::debug!("assignments of a `Schedule` should never be empty.");
+				1
+			};
 		let assignments = assignments
 			.into_iter()
 			.map(|(a, ratio)| (a, AssignmentState { ratio, remaining: ratio }))
@@ -149,7 +150,9 @@ pub mod pallet {
 	pub struct Pallet<T>(_);
 
 	#[pallet::config]
-	pub trait Config: frame_system::Config + configuration::Config + paras::Config + assigner_on_demand::Config {
+	pub trait Config:
+		frame_system::Config + configuration::Config + paras::Config + assigner_on_demand::Config
+	{
 		/// Something that provides the weight of this pallet.
 		type WeightInfo: WeightInfo;
 	}
@@ -253,8 +256,8 @@ impl<T: Config> AssignmentProvider<BlockNumberFor<T>> for Pallet<T> {
 					// advance:
 					a_state.remaining -= core_state.step;
 					if a_state.remaining < core_state.step {
-						// Assignment exhausted, need to move to the next and credit remaining for next
-						// round.
+						// Assignment exhausted, need to move to the next and credit remaining for
+						// next round.
 						core_state.pos += 1;
 						// Reset to ratio + still remaining "credits":
 						a_state.remaining += a_state.ratio;
@@ -267,7 +270,8 @@ impl<T: Config> AssignmentProvider<BlockNumberFor<T>> for Pallet<T> {
 								BlockNumberFor<T>,
 							>>::pop_assignment_for_core(core_idx)
 							.map(|assignment| BulkAssignment::Instantaneous(assignment)),
-						CoreAssignment::Task(para_id) => return Some(BulkAssignment::Bulk((*para_id).into())),
+						CoreAssignment::Task(para_id) =>
+							return Some(BulkAssignment::Bulk((*para_id).into())),
 					}
 				},
 				None => return None,
diff --git a/polkadot/runtime/parachains/src/assigner_parachains.rs b/polkadot/runtime/parachains/src/assigner_parachains.rs
index 3fa0ddd7a3dc..f7ae8ede98c4 100644
--- a/polkadot/runtime/parachains/src/assigner_parachains.rs
+++ b/polkadot/runtime/parachains/src/assigner_parachains.rs
@@ -101,4 +101,4 @@ impl<T: Config> AssignmentProvider<BlockNumberFor<T>> for Pallet<T> {
 			ttl: BlockNumberFor::<T>::from(10u32),
 		}
 	}
-}
\ No newline at end of file
+}
diff --git a/polkadot/runtime/parachains/src/assigner_parachains/tests.rs b/polkadot/runtime/parachains/src/assigner_parachains/tests.rs
index 5bdec939e61e..01053293554e 100644
--- a/polkadot/runtime/parachains/src/assigner_parachains/tests.rs
+++ b/polkadot/runtime/parachains/src/assigner_parachains/tests.rs
@@ -19,7 +19,9 @@ use super::*;
 use crate::{
 	assigner_parachains::{mock_helpers::GenesisConfigBuilder, ParachainsAssignment},
 	initializer::SessionChangeNotification,
-	mock::{new_test_ext, ParachainsAssigner, Paras, ParasShared, RuntimeOrigin, Scheduler, System},
+	mock::{
+		new_test_ext, ParachainsAssigner, Paras, ParasShared, RuntimeOrigin, Scheduler, System,
+	},
 	paras::{ParaGenesisArgs, ParaKind},
 };
 use frame_support::{assert_ok, pallet_prelude::*};
@@ -94,7 +96,8 @@ fn parachains_assigner_pop_assignment_is_always_some() {
 
 		for _ in 0..20 {
 			assert!(
-				ParachainsAssigner::pop_assignment_for_core(core_index) == Some(expected_assignment.clone())
+				ParachainsAssigner::pop_assignment_for_core(core_index) ==
+					Some(expected_assignment.clone())
 			);
 		}
 
@@ -102,7 +105,8 @@ fn parachains_assigner_pop_assignment_is_always_some() {
 
 		for _ in 0..20 {
 			assert!(
-				ParachainsAssigner::pop_assignment_for_core(core_index) == Some(expected_assignment.clone())
+				ParachainsAssigner::pop_assignment_for_core(core_index) ==
+					Some(expected_assignment.clone())
 			);
 		}
 	});

From 19de011832cc41f0c9f61ed6a05779e3a3e79b1d Mon Sep 17 00:00:00 2001
From: BradleyOlson64 <lotrftw9@gmail.com>
Date: Sat, 28 Oct 2023 13:11:35 -0700
Subject: [PATCH 04/10] Stub out some tests

---
 .../parachains/src/assigner_bulk/tests.rs     | 22 ++++++++++++++-----
 1 file changed, 16 insertions(+), 6 deletions(-)

diff --git a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
index b7d0da3a2073..372409de43b6 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
@@ -79,11 +79,21 @@ fn run_to_block(
 }
 
 #[test]
-fn place_order_works() {
-	let alice = 1u64;
-	let para_id = ParaId::from(111);
+fn pop_assignment_for_core_works() {
+	
+}
 
-	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
-		
-	});
+#[test]
+fn ensure_workload_works() {
+	
 }
+
+#[test]
+fn after_assign_core_workload_is_some() {
+
+}
+
+#[test]
+fn end_hint_always_points_to_next_work_plan_item() {
+
+}
\ No newline at end of file

From 34ac88d3dbd37f3a8c2f43f871dc9122bda31ecd Mon Sep 17 00:00:00 2001
From: BradleyOlson64 <lotrftw9@gmail.com>
Date: Mon, 30 Oct 2023 10:53:39 -0700
Subject: [PATCH 05/10] Fixed paras_inherent test

---
 polkadot/runtime/parachains/src/paras_inherent/tests.rs | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/polkadot/runtime/parachains/src/paras_inherent/tests.rs b/polkadot/runtime/parachains/src/paras_inherent/tests.rs
index 9f8cf0962874..6bef995d2cc4 100644
--- a/polkadot/runtime/parachains/src/paras_inherent/tests.rs
+++ b/polkadot/runtime/parachains/src/paras_inherent/tests.rs
@@ -666,12 +666,6 @@ mod enter {
 			// * 3 disputes.
 			assert_eq!(limit_inherent_data.disputes.len(), 2);
 
-			assert_ok!(Pallet::<Test>::enter(
-				frame_system::RawOrigin::None.into(),
-				limit_inherent_data,
-			));
-
-			// TODO [now]: this assertion fails with async backing runtime.
 			assert_eq!(
 				// The length of this vec is equal to the number of candidates, so we know 1
 				// candidate got filtered out

From a90535b11a97fff1dc52cc919c28134dced7b264 Mon Sep 17 00:00:00 2001
From: BradleyOlson64 <lotrftw9@gmail.com>
Date: Mon, 30 Oct 2023 11:47:50 -0700
Subject: [PATCH 06/10] Making workplan and workload value query

---
 .../parachains/src/assigner_bulk/mod.rs       | 87 ++++++++++---------
 1 file changed, 48 insertions(+), 39 deletions(-)

diff --git a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
index eb406a803aba..85ebbb1e9fff 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
@@ -79,6 +79,16 @@ struct Schedule<N> {
 	end_hint: Option<N>,
 }
 
+// TODO: end_hint should probably be chosen in a smarter way.
+impl<N> Default for Schedule<N> {
+	fn default() -> Self {
+		Schedule {
+			assignments: vec![(CoreAssignment::Pool, PartsOf57600::from(57600u16))],
+			end_hint: None
+		}
+	}
+}
+
 /// An instantiated `Schedule`.
 ///
 /// This is the state of assignments currently being served via the `AssignmentProvider` interface,
@@ -110,6 +120,12 @@ struct CoreState<N> {
 	step: PartsOf57600,
 }
 
+impl<N> Default for CoreState<N> {
+	fn default() -> Self {
+		CoreState::from(Schedule::default())
+	}
+}
+
 #[derive(Encode, Decode, TypeInfo)]
 struct AssignmentState {
 	/// Ratio of the core this assignment has.
@@ -175,7 +191,8 @@ pub mod pallet {
 		Twox256,
 		(BlockNumberFor<T>, CoreIndex),
 		Schedule<BlockNumberFor<T>>,
-		OptionQuery,
+		ValueQuery,
+		GetDefault,
 	>;
 
 	/// Latest schedule for the given core.
@@ -191,7 +208,7 @@ pub mod pallet {
 	/// `PendingAssignments`.
 	#[pallet::storage]
 	pub(super) type Workload<T: Config> =
-		StorageMap<_, Twox256, CoreIndex, CoreState<BlockNumberFor<T>>, OptionQuery>;
+		StorageMap<_, Twox256, CoreIndex, CoreState<BlockNumberFor<T>>, ValueQuery, GetDefault>;
 
 	#[pallet::hooks]
 	impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
@@ -249,36 +266,32 @@ impl<T: Config> AssignmentProvider<BlockNumberFor<T>> for Pallet<T> {
 
 		Workload::<T>::mutate(core_idx, |core_state| {
 			Self::ensure_workload(now, core_idx, core_state);
-			match core_state {
-				Some(core_state) => {
-					core_state.pos = core_state.pos % core_state.assignments.len() as u16;
-					let (a_type, a_state) = &mut core_state
-						.assignments
-						.get_mut(core_state.pos as usize)
-						.expect("We limited pos to the size of the vec one line above. qed");
-
-					// advance:
-					a_state.remaining -= core_state.step;
-					if a_state.remaining < core_state.step {
-						// Assignment exhausted, need to move to the next and credit remaining for
-						// next round.
-						core_state.pos += 1;
-						// Reset to ratio + still remaining "credits":
-						a_state.remaining += a_state.ratio;
-					}
-
-					match a_type {
-						CoreAssignment::Idle => return None,
-						CoreAssignment::Pool =>
-							return <assigner_on_demand::Pallet<T> as AssignmentProvider<
-								BlockNumberFor<T>,
-							>>::pop_assignment_for_core(core_idx)
-							.map(|assignment| BulkAssignment::Instantaneous(assignment)),
-						CoreAssignment::Task(para_id) =>
-							return Some(BulkAssignment::Bulk((*para_id).into())),
-					}
-				},
-				None => return None,
+
+			core_state.pos = core_state.pos % core_state.assignments.len() as u16;
+			let (a_type, a_state) = &mut core_state
+				.assignments
+				.get_mut(core_state.pos as usize)
+				.expect("We limited pos to the size of the vec one line above. qed");
+
+			// advance:
+			a_state.remaining -= core_state.step;
+			if a_state.remaining < core_state.step {
+				// Assignment exhausted, need to move to the next and credit remaining for
+				// next round.
+				core_state.pos += 1;
+				// Reset to ratio + still remaining "credits":
+				a_state.remaining += a_state.ratio;
+			}
+
+			match a_type {
+				CoreAssignment::Idle => return None,
+				CoreAssignment::Pool =>
+					return <assigner_on_demand::Pallet<T> as AssignmentProvider<
+						BlockNumberFor<T>,
+					>>::pop_assignment_for_core(core_idx)
+					.map(|assignment| BulkAssignment::Instantaneous(assignment)),
+				CoreAssignment::Task(para_id) =>
+					return Some(BulkAssignment::Bulk((*para_id).into())),
 			}
 		})
 	}
@@ -320,9 +333,9 @@ impl<T: Config> Pallet<T> {
 	fn ensure_workload(
 		now: BlockNumberFor<T>,
 		core_idx: CoreIndex,
-		workload: &mut Option<CoreState<BlockNumberFor<T>>>,
+		workload: &mut CoreState<BlockNumberFor<T>>,
 	) {
-		let update = if let Some(workload) = workload {
+		let update = {
 			match workload.end_hint {
 				Some(end_hint) if end_hint < now => {
 					// Invariant: Always points to next item in `Workplan`, if such an item exists.
@@ -335,14 +348,10 @@ impl<T: Config> Pallet<T> {
 				// No end in sight, still valid:
 				None => return,
 			}
-		} else {
-			// Invariant: If there is no workload, workplan must be empty for core.
-			// Therefore nothing to do here.
-			return
 		};
 
 		// Needs update:
-		*workload = update.map(|schedule| schedule.into());
+		*workload = update.into();
 	}
 }
 

From c7b053caf37378ba44c762a0fd7a781877c8aba6 Mon Sep 17 00:00:00 2001
From: BradleyOlson64 <lotrftw9@gmail.com>
Date: Mon, 30 Oct 2023 11:56:04 -0700
Subject: [PATCH 07/10] fmt

---
 .../runtime/parachains/src/assigner_bulk/mod.rs  |  2 +-
 .../parachains/src/assigner_bulk/tests.rs        | 16 ++++------------
 2 files changed, 5 insertions(+), 13 deletions(-)

diff --git a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
index 85ebbb1e9fff..bae9212b26c9 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
@@ -84,7 +84,7 @@ impl<N> Default for Schedule<N> {
 	fn default() -> Self {
 		Schedule {
 			assignments: vec![(CoreAssignment::Pool, PartsOf57600::from(57600u16))],
-			end_hint: None
+			end_hint: None,
 		}
 	}
 }
diff --git a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
index 372409de43b6..c0ab505377de 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
@@ -79,21 +79,13 @@ fn run_to_block(
 }
 
 #[test]
-fn pop_assignment_for_core_works() {
-	
-}
+fn pop_assignment_for_core_works() {}
 
 #[test]
-fn ensure_workload_works() {
-	
-}
+fn ensure_workload_works() {}
 
 #[test]
-fn after_assign_core_workload_is_some() {
-
-}
+fn after_assign_core_workload_is_some() {}
 
 #[test]
-fn end_hint_always_points_to_next_work_plan_item() {
-
-}
\ No newline at end of file
+fn end_hint_always_points_to_next_work_plan_item() {}

From d7e8081e10b0f4cba3b6b9589f583f650e5f18a6 Mon Sep 17 00:00:00 2001
From: BradleyOlson64 <lotrftw9@gmail.com>
Date: Mon, 30 Oct 2023 14:49:21 -0700
Subject: [PATCH 08/10] Stub out assign_core_test

---
 polkadot/runtime/parachains/src/assigner_bulk/tests.rs | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
index c0ab505377de..e54c02cbd0a3 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
@@ -89,3 +89,9 @@ fn after_assign_core_workload_is_some() {}
 
 #[test]
 fn end_hint_always_points_to_next_work_plan_item() {}
+
+// Invariants: We assume that Workplan is append only and consumed. In other words new schedules
+// inserted for a core must have a higher block number than all of the already existing
+// schedules.
+#[test]
+fn assign_core_enforces_higher_block_number() {}

From 583d1473a5d0916bb8fcac9e55f5141a4a8b573b Mon Sep 17 00:00:00 2001
From: BradleyOlson64 <lotrftw9@gmail.com>
Date: Fri, 3 Nov 2023 12:50:44 -0700
Subject: [PATCH 09/10] Stubbed out assign_core

---
 .../parachains/src/assigner_bulk/mod.rs       | 39 +++++++++++++++++++
 .../parachains/src/assigner_bulk/tests.rs     |  6 ++-
 2 files changed, 44 insertions(+), 1 deletion(-)

diff --git a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
index bae9212b26c9..c6addff467d1 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
@@ -219,6 +219,12 @@ pub mod pallet {
 
 	#[pallet::call]
 	impl<T: Config> Pallet<T> {}
+
+	#[pallet::error]
+	pub enum Error<T> {
+		// Attempted to assign an invalid schedule to a core in WorkPlan
+		InvalidScheduleAssigned,
+	}
 }
 
 /// Assignments as provided by our `AssignmentProvider` implementation.
@@ -353,6 +359,39 @@ impl<T: Config> Pallet<T> {
 		// Needs update:
 		*workload = update.into();
 	}
+
+	// Add a schedule to the given core which starts at the given block number. 
+	// This schedule of work will be serviced indefinitely unless some new 
+	// schedule is added.
+	fn assign_core(
+		schedule: Schedule<T>,
+		core_idx: CoreIndex,
+		schedule_start: BlockNumberFor::<T>,
+	) -> Result<(), DispatchError> {
+		// TODO: Make sure this check is appropriate based on when and how we assign
+		// schedule endings
+		ensure!(schedule.end_hint.is_none(), Error::<T>::InvalidScheduleAssigned);
+
+		// Check that the total parts between all assignments do not exceed 57600
+		ensure!(schedule.assignments.iter().map(|assignment| assignment.1).fold(PartsOf57600::from(0u16), |sum, parts| sum.saturating_add(parts)) 
+				<= PartsOf57600::from(57600u16), Error::<T>::InvalidScheduleAssigned);
+
+		// Check that schedule_start is a higher block number than that for any other
+		// schedule for the same core
+		//ensure!(Workplan::<T>::get(core_idx).last(),
+				//Error::<T>::InvalidScheduleAssigned);
+
+		// TODO: Possibly check that schedule start isn't in the past
+
+		// Enter schedule into workplan
+
+		// Add end_hint to prior schedule, if any
+
+		// If the workplan for this core is empty but there is an active workload,
+		// set an end_hint for the active workload
+
+		Ok(())
+	}
 }
 
 // Tests/Invariant:
diff --git a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
index e54c02cbd0a3..d207b942033d 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
@@ -90,8 +90,12 @@ fn after_assign_core_workload_is_some() {}
 #[test]
 fn end_hint_always_points_to_next_work_plan_item() {}
 
+#[test]
 // Invariants: We assume that Workplan is append only and consumed. In other words new schedules
 // inserted for a core must have a higher block number than all of the already existing
 // schedules.
-#[test]
 fn assign_core_enforces_higher_block_number() {}
+
+#[test]
+// A schedule is poorly formed if it has an end_hint from 
+fn assign_core_enforces_well_formed_schedule() {}

From 1027244fe8403ee942917b8439bf6bc85ee86d2a Mon Sep 17 00:00:00 2001
From: BradleyOlson64 <lotrftw9@gmail.com>
Date: Tue, 7 Nov 2023 13:25:46 -0800
Subject: [PATCH 10/10] Filled in add_assignment and bulk assigner tests

---
 .../src/assigner_bulk/mock_helpers.rs         |   1 +
 .../parachains/src/assigner_bulk/mod.rs       | 103 +++--
 .../parachains/src/assigner_bulk/tests.rs     | 400 +++++++++++++++++-
 polkadot/runtime/parachains/src/mock.rs       |   8 +-
 4 files changed, 458 insertions(+), 54 deletions(-)

diff --git a/polkadot/runtime/parachains/src/assigner_bulk/mock_helpers.rs b/polkadot/runtime/parachains/src/assigner_bulk/mock_helpers.rs
index acfb24cbf194..3703bf0e6e99 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/mock_helpers.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/mock_helpers.rs
@@ -24,6 +24,7 @@ use crate::{
 	mock::MockGenesisConfig,
 	paras::{ParaGenesisArgs, ParaKind},
 };
+use sp_runtime::Perbill;
 
 use primitives::{Balance, HeadData, ValidationCode};
 
diff --git a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
index c6addff467d1..32cd4400b22f 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/mod.rs
@@ -29,23 +29,13 @@ use crate::{
 	},
 };
 
-use frame_support::{
-	pallet_prelude::*,
-	traits::{
-		Currency,
-		ExistenceRequirement::{self, AllowDeath, KeepAlive},
-		WithdrawReasons,
-	},
-};
+use frame_support::pallet_prelude::*;
 use frame_system::pallet_prelude::*;
 use pallet_broker::CoreAssignment;
 use primitives::{CoreIndex, Id as ParaId};
-use sp_runtime::{
-	traits::{One, SaturatedConversion},
-	FixedPointNumber, FixedPointOperand, FixedU128, Perbill, Saturating,
-};
+use sp_runtime::Saturating;
 
-use sp_std::{collections::vec_deque::VecDeque, prelude::*};
+use sp_std::prelude::*;
 
 const LOG_TARGET: &str = "runtime::parachains::assigner-bulk";
 
@@ -65,6 +55,7 @@ impl WeightInfo for TestWeightInfo {}
 ///
 /// for a particular core.
 #[derive(Encode, Decode, TypeInfo)]
+#[cfg_attr(test, derive(PartialEq, RuntimeDebug))]
 struct Schedule<N> {
 	// Original assignments
 	assignments: Vec<(CoreAssignment, PartsOf57600)>,
@@ -94,6 +85,7 @@ impl<N> Default for Schedule<N> {
 /// This is the state of assignments currently being served via the `AssignmentProvider` interface,
 /// as opposed to `Schedule` which is upcoming not yet served assignments.
 #[derive(Encode, Decode, TypeInfo)]
+#[cfg_attr(test, derive(PartialEq, RuntimeDebug, Clone))]
 struct CoreState<N> {
 	/// Assignments with current state.
 	///
@@ -127,6 +119,7 @@ impl<N> Default for CoreState<N> {
 }
 
 #[derive(Encode, Decode, TypeInfo)]
+#[cfg_attr(test, derive(PartialEq, RuntimeDebug, Clone, Copy))]
 struct AssignmentState {
 	/// Ratio of the core this assignment has.
 	///
@@ -191,8 +184,7 @@ pub mod pallet {
 		Twox256,
 		(BlockNumberFor<T>, CoreIndex),
 		Schedule<BlockNumberFor<T>>,
-		ValueQuery,
-		GetDefault,
+		OptionQuery,
 	>;
 
 	/// Latest schedule for the given core.
@@ -213,7 +205,8 @@ pub mod pallet {
 	#[pallet::hooks]
 	impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {
 		fn on_initialize(_now: BlockNumberFor<T>) -> Weight {
-			unimplemented!("TODO: Implement")
+			//TODO: Implement
+			T::DbWeight::get().reads_writes(0, 0)
 		}
 	}
 
@@ -229,6 +222,7 @@ pub mod pallet {
 
 /// Assignments as provided by our `AssignmentProvider` implementation.
 #[derive(Encode, Decode, TypeInfo, Debug)]
+#[cfg_attr(test, derive(PartialEq))]
 pub enum BulkAssignment<OnDemand> {
 	/// Assignment was an instantaneous core time assignment.
 	Instantaneous(OnDemand),
@@ -258,7 +252,7 @@ impl<T: Config> AssignmentProvider<BlockNumberFor<T>> for Pallet<T> {
 
 	fn migrate_old_to_current(
 		old: Self::OldAssignmentType,
-		core: CoreIndex,
+		_core: CoreIndex,
 	) -> Self::AssignmentType {
 		old
 	}
@@ -345,9 +339,13 @@ impl<T: Config> Pallet<T> {
 			match workload.end_hint {
 				Some(end_hint) if end_hint < now => {
 					// Invariant: Always points to next item in `Workplan`, if such an item exists.
-					let n = end_hint.saturating_add(BlockNumberFor::<T>::from(1u32));
-					// Workload expired - update to whatever is scheduled or `None` if nothing is:
-					Workplan::<T>::take((n, core_idx))
+					let n = end_hint.saturating_plus_one();
+					// Workload expired - update to whatever is scheduled, or a default workload if
+					// nothing is:
+					match Workplan::<T>::take((n, core_idx)) {
+						Some(schedule) => schedule,
+						None => Schedule::default(),
+					}
 				},
 				// Still alive:
 				Some(_) => return,
@@ -360,36 +358,63 @@ impl<T: Config> Pallet<T> {
 		*workload = update.into();
 	}
 
-	// Add a schedule to the given core which starts at the given block number. 
-	// This schedule of work will be serviced indefinitely unless some new 
+	// Add a schedule to the given core which starts at the given block number.
+	// This schedule of work will be serviced indefinitely unless some new
 	// schedule is added.
 	fn assign_core(
-		schedule: Schedule<T>,
+		schedule: Schedule<BlockNumberFor<T>>,
 		core_idx: CoreIndex,
-		schedule_start: BlockNumberFor::<T>,
+		schedule_start: BlockNumberFor<T>,
 	) -> Result<(), DispatchError> {
 		// TODO: Make sure this check is appropriate based on when and how we assign
 		// schedule endings
 		ensure!(schedule.end_hint.is_none(), Error::<T>::InvalidScheduleAssigned);
 
-		// Check that the total parts between all assignments do not exceed 57600
-		ensure!(schedule.assignments.iter().map(|assignment| assignment.1).fold(PartsOf57600::from(0u16), |sum, parts| sum.saturating_add(parts)) 
-				<= PartsOf57600::from(57600u16), Error::<T>::InvalidScheduleAssigned);
+		// There should be at least one assignment
+		ensure!(schedule.assignments.len() > 0usize, Error::<T>::InvalidScheduleAssigned);
 
-		// Check that schedule_start is a higher block number than that for any other
-		// schedule for the same core
-		//ensure!(Workplan::<T>::get(core_idx).last(),
-				//Error::<T>::InvalidScheduleAssigned);
-
-		// TODO: Possibly check that schedule start isn't in the past
+		// Check that the total parts between all assignments do not exceed 57600
+		ensure!(
+			schedule
+				.assignments
+				.iter()
+				.map(|assignment| assignment.1)
+				.fold(PartsOf57600::from(0u16), |sum, parts| sum.saturating_add(parts)) <=
+				PartsOf57600::from(57600u16),
+			Error::<T>::InvalidScheduleAssigned
+		);
+
+		let mut prior_schedule_start: Option<BlockNumberFor<T>> = None;
+		LatestCoreSchedule::<T>::try_mutate(core_idx, |maybe_prior_schedule| {
+			prior_schedule_start = maybe_prior_schedule.clone();
+			// Check that schedule_start is a higher block number than that of any other
+			// schedule for the same core
+			if let Some(prior_schedule) = maybe_prior_schedule {
+				ensure!(*prior_schedule < schedule_start, Error::<T>::InvalidScheduleAssigned);
+			}
+			*maybe_prior_schedule = Some(schedule_start);
+			Ok::<(), Error<T>>(())
+		})?;
 
 		// Enter schedule into workplan
-
-		// Add end_hint to prior schedule, if any
-
-		// If the workplan for this core is empty but there is an active workload,
-		// set an end_hint for the active workload
-
+		Workplan::<T>::insert((schedule_start, core_idx), schedule);
+
+		// Add end_hint to prior schedule, if any. Else, set end hint for this core's
+		// CoreState. This call is why using OptionQuery for Workplan is helpful.
+		let mut has_prior_schedule = false;
+		if let Some(prior_schedule_start) = prior_schedule_start {
+			Workplan::<T>::mutate((prior_schedule_start, core_idx), |maybe_prior_schedule| {
+				if let Some(prior_schedule) = maybe_prior_schedule {
+					prior_schedule.end_hint = Some(schedule_start.saturating_less_one());
+					has_prior_schedule = true;
+				}
+			});
+		}
+		if has_prior_schedule == false {
+			Workload::<T>::mutate(core_idx, |core_state| {
+				core_state.end_hint = Some(schedule_start.saturating_less_one());
+			});
+		}
 		Ok(())
 	}
 }
diff --git a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
index d207b942033d..f14baa0ce067 100644
--- a/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
+++ b/polkadot/runtime/parachains/src/assigner_bulk/tests.rs
@@ -17,16 +17,17 @@
 use super::*;
 
 use crate::{
-	assigner_bulk::mock_helpers::GenesisConfigBuilder,
+	assigner_bulk::{mock_helpers::GenesisConfigBuilder, pallet::Error, Schedule},
+	assigner_on_demand::OnDemandAssignment,
 	initializer::SessionChangeNotification,
 	mock::{
-		new_test_ext, Balances, OnDemandAssigner, Paras, ParasShared, RuntimeOrigin, Scheduler,
-		System, Test,
+		new_test_ext, Balances, BulkAssigner, OnDemandAssigner, Paras, ParasShared, RuntimeOrigin,
+		Scheduler, System, Test,
 	},
 	paras::{ParaGenesisArgs, ParaKind},
 };
-use frame_support::{assert_noop, assert_ok, error::BadOrigin};
-use pallet_balances::Error as BalancesError;
+use frame_support::{assert_noop, assert_ok, pallet_prelude::*, traits::Currency};
+use pallet_broker::TaskId;
 use primitives::{BlockNumber, SessionIndex, ValidationCode};
 use sp_std::collections::btree_map::BTreeMap;
 
@@ -79,23 +80,394 @@ fn run_to_block(
 }
 
 #[test]
-fn pop_assignment_for_core_works() {}
+// Should update end hint of current workload and add new schedule to
+// Workplan
+fn assign_core_works_with_no_prior_schedule() {
+	let core_idx = CoreIndex(0);
 
-#[test]
-fn ensure_workload_works() {}
+	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+		run_to_block(10, |n| if n == 10 { Some(Default::default()) } else { None });
 
-#[test]
-fn after_assign_core_workload_is_some() {}
+		// Call assign_core
+		assert_ok!(BulkAssigner::assign_core(
+			Schedule::default(),
+			core_idx,
+			BlockNumberFor::<Test>::from(11u32),
+		));
+
+		// Check Workplan
+		assert_eq!(
+			Workplan::<Test>::get((BlockNumberFor::<Test>::from(11u32), core_idx)),
+			Some(Schedule::default())
+		);
+
+		// Check CoreState end_hint
+		assert_eq!(
+			Workload::<Test>::get(core_idx).end_hint,
+			Some(BlockNumberFor::<Test>::from(10u32))
+		);
+	});
+}
 
 #[test]
-fn end_hint_always_points_to_next_work_plan_item() {}
+// Should update the end hint of prior schedule and add new schedule
+// to Workplan
+fn assign_core_works_with_prior_schedule() {
+	let core_idx = CoreIndex(0);
+
+	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+		run_to_block(10, |n| if n == 10 { Some(Default::default()) } else { None });
+		let default_with_end_hint = Schedule { end_hint: Some(14u32), ..Schedule::default() };
+
+		// Call assign_core twice
+		assert_ok!(BulkAssigner::assign_core(
+			Schedule::default(),
+			core_idx,
+			BlockNumberFor::<Test>::from(11u32),
+		));
+
+		assert_ok!(BulkAssigner::assign_core(
+			Schedule::default(),
+			core_idx,
+			BlockNumberFor::<Test>::from(15u32),
+		));
+
+		// Check CoreState end_hint
+		assert_eq!(
+			Workload::<Test>::get(core_idx).end_hint,
+			Some(BlockNumberFor::<Test>::from(10u32))
+		);
+
+		// Check Workplan for two entries
+		assert_eq!(
+			Workplan::<Test>::get((BlockNumberFor::<Test>::from(11u32), core_idx)),
+			Some(default_with_end_hint)
+		);
+		assert_eq!(
+			Workplan::<Test>::get((BlockNumberFor::<Test>::from(15u32), core_idx)),
+			Some(Schedule::default())
+		);
+	});
+}
 
 #[test]
 // Invariants: We assume that Workplan is append only and consumed. In other words new schedules
 // inserted for a core must have a higher block number than all of the already existing
 // schedules.
-fn assign_core_enforces_higher_block_number() {}
+fn assign_core_enforces_higher_block_number() {
+	let core_idx = CoreIndex(0);
+
+	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+		run_to_block(10, |n| if n == 10 { Some(Default::default()) } else { None });
+
+		// Call assign core once with higher starting block number
+		assert_ok!(BulkAssigner::assign_core(
+			Schedule::default(),
+			core_idx,
+			BlockNumberFor::<Test>::from(11u32),
+		));
+
+		// Call again with lower starting block number, expecting an error
+		assert_noop!(
+			BulkAssigner::assign_core(
+				Schedule::default(),
+				core_idx,
+				BlockNumberFor::<Test>::from(10u32),
+			),
+			Error::<Test>::InvalidScheduleAssigned
+		);
+	});
+}
+
+#[test]
+fn assign_core_enforces_well_formed_schedule() {
+	let para_id = ParaId::from(1u32);
+	let core_idx = CoreIndex(0);
+
+	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+		run_to_block(10, |n| if n == 10 { Some(Default::default()) } else { None });
+
+		// Making invalid schedules
+		let bad_end_hint = Schedule { end_hint: Some(20), ..Schedule::default() };
+		let bad_assignment_count = Schedule { assignments: vec![], ..Schedule::default() };
+		let bad_parts_sum = Schedule {
+			assignments: vec![
+				(CoreAssignment::Task(para_id.into()), PartsOf57600::from(57600u16)),
+				(CoreAssignment::Pool, PartsOf57600::from(57600u16)),
+			],
+			..Schedule::default()
+		};
+
+		// Attempting to assign_core with bad schedules
+		assert_noop!(
+			BulkAssigner::assign_core(bad_end_hint, core_idx, BlockNumberFor::<Test>::from(11u32),),
+			Error::<Test>::InvalidScheduleAssigned
+		);
+		assert_noop!(
+			BulkAssigner::assign_core(
+				bad_assignment_count,
+				core_idx,
+				BlockNumberFor::<Test>::from(11u32),
+			),
+			Error::<Test>::InvalidScheduleAssigned
+		);
+		assert_noop!(
+			BulkAssigner::assign_core(bad_parts_sum, core_idx, BlockNumberFor::<Test>::from(11u32),),
+			Error::<Test>::InvalidScheduleAssigned
+		);
+	});
+}
+
+#[test]
+fn end_hint_always_points_to_next_work_plan_item() {
+	let core_idx = CoreIndex(0);
+
+	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+		run_to_block(10, |n| if n == 10 { Some(Default::default()) } else { None });
+		let start_1 = 10u32;
+		let start_2 = 15u32;
+		let start_3 = 20u32;
+		let start_4 = 25u32;
+		let start_5 = 30u32;
+
+		let expected_core_state_3 =
+			CoreState { end_hint: Some(start_4 - 1), pos: 1u16, ..CoreState::default() };
+		let expected_schedule_4 = Schedule { end_hint: Some(start_5 - 1), ..Schedule::default() };
+		let expected_schedule_5 = Schedule::default();
+
+		// Call assign_core for each of five schedules
+		assert_ok!(BulkAssigner::assign_core(
+			Schedule::default(),
+			core_idx,
+			BlockNumberFor::<Test>::from(start_1),
+		));
+
+		assert_ok!(BulkAssigner::assign_core(
+			Schedule::default(),
+			core_idx,
+			BlockNumberFor::<Test>::from(start_2),
+		));
+
+		assert_ok!(BulkAssigner::assign_core(
+			Schedule::default(),
+			core_idx,
+			BlockNumberFor::<Test>::from(start_3),
+		));
+
+		assert_ok!(BulkAssigner::assign_core(
+			Schedule::default(),
+			core_idx,
+			BlockNumberFor::<Test>::from(start_4),
+		));
+
+		assert_ok!(BulkAssigner::assign_core(
+			Schedule::default(),
+			core_idx,
+			BlockNumberFor::<Test>::from(start_5),
+		));
+
+		// Rotate through the first three schedules
+		BulkAssigner::pop_assignment_for_core(core_idx);
+		run_to_block(15, |n| if n == 15 { Some(Default::default()) } else { None });
+		BulkAssigner::pop_assignment_for_core(core_idx);
+		run_to_block(20, |n| if n == 20 { Some(Default::default()) } else { None });
+		BulkAssigner::pop_assignment_for_core(core_idx);
+
+		// Use saved starting block numbers to check that schedules chain
+		// together correctly
+		assert_eq!(Workload::<Test>::get(core_idx), expected_core_state_3);
+
+		assert_eq!(
+			Workplan::<Test>::get((BlockNumberFor::<Test>::from(start_4), core_idx)),
+			Some(expected_schedule_4)
+		);
+		assert_eq!(
+			Workplan::<Test>::get((BlockNumberFor::<Test>::from(start_5), core_idx)),
+			Some(expected_schedule_5)
+		);
+	});
+}
+
+#[test]
+fn ensure_workload_works() {
+	let core_idx = CoreIndex(0);
+	let task_1 = TaskId::from(1u32);
+	let task_2 = TaskId::from(2u32);
+	let test_assignment_state = AssignmentState {
+		ratio: PartsOf57600::from(57600u16),
+		remaining: PartsOf57600::from(57600u16),
+	};
+
+	let mut workload: CoreState<BlockNumberFor<Test>> = CoreState {
+		assignments: vec![(CoreAssignment::Task(task_1), test_assignment_state)],
+		..CoreState::default()
+	};
+	let after_case_1 = CoreState { ..workload.clone() };
+	let after_case_2 =
+		CoreState { end_hint: Some(BlockNumberFor::<Test>::from(10u32)), ..workload.clone() };
+	let after_case_3 = CoreState {
+		assignments: vec![(CoreAssignment::Task(task_2), test_assignment_state)],
+		..CoreState::default()
+	};
+
+	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+		run_to_block(10, |n| if n == 10 { Some(Default::default()) } else { None });
+
+		// Case 1: No schedules in workplan for core
+		BulkAssigner::ensure_workload(10u32, core_idx, &mut workload);
+
+		assert_eq!(workload, after_case_1);
+
+		// Case 2: Schedule in workplan for core, but end_hint not reached
+		let schedule: Schedule<BlockNumberFor<Test>> = Schedule {
+			assignments: vec![(CoreAssignment::Task(task_2), PartsOf57600::from(57600u16))],
+			..Schedule::default()
+		};
+
+		assert_ok!(BulkAssigner::assign_core(
+			schedule,
+			core_idx,
+			BlockNumberFor::<Test>::from(11u32),
+		));
+
+		// Propagate end_hint modification due to assign_core from CoreState in
+		// storage to local core state. Normally pop_assignment_for_core would
+		// handle this.
+		workload.end_hint = Workload::<Test>::get(core_idx).end_hint;
+
+		BulkAssigner::ensure_workload(10u32, core_idx, &mut workload);
+
+		assert_eq!(workload, after_case_2);
+
+		// Case 3: Schedule in workplan for core, end_hint reached. Swaps new CoreState
+		// into Workload from Workplan.
+		BulkAssigner::ensure_workload(11u32, core_idx, &mut workload);
+
+		assert_eq!(workload, after_case_3);
+	});
+}
 
 #[test]
-// A schedule is poorly formed if it has an end_hint from 
-fn assign_core_enforces_well_formed_schedule() {}
+fn pop_assignment_for_core_works() {
+	let para_id = ParaId::from(1);
+	let core_idx = CoreIndex(0);
+	let alice = 1u64;
+	let amt = 10_000_000u128;
+	let on_demand_assignment = OnDemandAssignment::new(para_id, CoreIndex(0));
+
+	let schedule_idle: Schedule<BlockNumberFor<Test>> = Schedule {
+		assignments: vec![(CoreAssignment::Idle, PartsOf57600::from(57600u16))],
+		..Schedule::default()
+	};
+	let schedule_task: Schedule<BlockNumberFor<Test>> = Schedule {
+		assignments: vec![(CoreAssignment::Task(para_id.into()), PartsOf57600::from(57600u16))],
+		..Schedule::default()
+	};
+
+	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+		// Initialize the parathread, wait for it to be ready, then add an
+		// on demand order to later pop with our bulk assigner.
+		schedule_blank_para(para_id, ParaKind::Parathread);
+		Balances::make_free_balance_be(&alice, amt);
+		run_to_block(10, |n| if n == 10 { Some(Default::default()) } else { None });
+		assert_ok!(OnDemandAssigner::place_order_allow_death(
+			RuntimeOrigin::signed(alice),
+			amt,
+			para_id
+		));
+
+		// Case 1: Assignment idle
+		assert_ok!(BulkAssigner::assign_core(
+			schedule_idle,
+			core_idx,
+			BlockNumberFor::<Test>::from(10u32),
+		));
+
+		assert_eq!(BulkAssigner::pop_assignment_for_core(core_idx), None);
+
+		// Case 2: Assignment pool
+		assert_ok!(BulkAssigner::assign_core(
+			Schedule::default(), //Default is pool
+			core_idx,
+			BlockNumberFor::<Test>::from(11u32),
+		));
+
+		run_to_block(11, |n| if n == 11 { Some(Default::default()) } else { None });
+
+		assert_eq!(
+			BulkAssigner::pop_assignment_for_core(core_idx),
+			Some(BulkAssignment::Instantaneous(on_demand_assignment))
+		);
+
+		// Case 3: Assignment task
+		assert_ok!(BulkAssigner::assign_core(
+			schedule_task,
+			core_idx,
+			BlockNumberFor::<Test>::from(12u32),
+		));
+
+		run_to_block(12, |n| if n == 12 { Some(Default::default()) } else { None });
+
+		assert_eq!(
+			BulkAssigner::pop_assignment_for_core(core_idx),
+			Some(BulkAssignment::Bulk(para_id))
+		);
+	});
+}
+
+#[test]
+fn assignment_proportions_in_core_state_work() {
+	let core_idx = CoreIndex(0);
+	let task_1 = TaskId::from(1u32);
+	let task_2 = TaskId::from(2u32);
+
+	new_test_ext(GenesisConfigBuilder::default().build()).execute_with(|| {
+		run_to_block(10, |n| if n == 10 { Some(Default::default()) } else { None });
+
+		// Task 1 gets 2/3 core usage, while task 2 gets 1/3
+		let test_schedule = Schedule {
+			assignments: vec![
+				(CoreAssignment::Task(task_1), PartsOf57600::from(57600u16 / 3 * 2)),
+				(CoreAssignment::Task(task_2), PartsOf57600::from(57600u16 / 3)),
+			],
+			..Schedule::default()
+		};
+
+		assert_ok!(BulkAssigner::assign_core(
+			test_schedule,
+			core_idx,
+			BlockNumberFor::<Test>::from(10u32),
+		));
+
+		// Case 1: Current assignment remaining >= step after pop
+		{
+			assert_eq!(
+				BulkAssigner::pop_assignment_for_core(core_idx),
+				Some(BulkAssignment::Bulk(task_1.into()))
+			);
+
+			assert_eq!(Workload::<Test>::get(core_idx).pos, 0u16);
+			// Consumed step should be 1/3 of core parts, leaving 1/3 remaining
+			assert_eq!(
+				Workload::<Test>::get(core_idx).assignments[0].1.remaining,
+				PartsOf57600::from(57600u16 / 3)
+			);
+		}
+
+		// Case 2: Current assignment remaning < step after pop
+		{
+			assert_eq!(
+				BulkAssigner::pop_assignment_for_core(core_idx),
+				Some(BulkAssignment::Bulk(task_1.into()))
+			);
+			// Pos should have incremented, as assignment had remaining < step
+			assert_eq!(Workload::<Test>::get(core_idx).pos, 1u16);
+			// Remaining should have started at 1/3 of core work parts. We then subtract
+			// step (1/3) and add back ratio (2/3), leaving us with 2/3 of core work parts.
+			assert_eq!(
+				Workload::<Test>::get(core_idx).assignments[0].1.remaining,
+				PartsOf57600::from(57600u16 / 3 * 2)
+			);
+		}
+	});
+}
diff --git a/polkadot/runtime/parachains/src/mock.rs b/polkadot/runtime/parachains/src/mock.rs
index f913646fb6bf..b3091f4c4543 100644
--- a/polkadot/runtime/parachains/src/mock.rs
+++ b/polkadot/runtime/parachains/src/mock.rs
@@ -17,7 +17,8 @@
 //! Mocks for all the traits.
 
 use crate::{
-	assigner, assigner_on_demand, assigner_parachains, configuration, disputes, dmp, hrmp,
+	assigner, assigner_bulk, assigner_on_demand, assigner_parachains, configuration, disputes, dmp,
+	hrmp,
 	inclusion::{self, AggregateMessageOrigin, UmpQueueId},
 	initializer, origin, paras,
 	paras::ParaKind,
@@ -73,6 +74,7 @@ frame_support::construct_runtime!(
 		Assigner: assigner,
 		ParachainsAssigner: assigner_parachains,
 		OnDemandAssigner: assigner_on_demand,
+		BulkAssigner: assigner_bulk,
 		Initializer: initializer,
 		Dmp: dmp,
 		Hrmp: hrmp,
@@ -363,6 +365,10 @@ impl assigner_on_demand::Config for Test {
 	type WeightInfo = crate::assigner_on_demand::TestWeightInfo;
 }
 
+impl assigner_bulk::Config for Test {
+	type WeightInfo = crate::assigner_bulk::TestWeightInfo;
+}
+
 impl crate::inclusion::Config for Test {
 	type WeightInfo = ();
 	type RuntimeEvent = RuntimeEvent;