// This file is part of Substrate.

// Copyright (C) Parity Technologies (UK) Ltd.

// SPDX-License-Identifier: Apache-2.0

// Licensed under the Apache License, Version 2.0 (the "License");

// you may not use this file except in compliance with the License.

// You may obtain a copy of the License at

//

// 	http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software

// distributed under the License is distributed on an "AS IS" BASIS,

// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

// See the License for the specific language governing permissions and

// limitations under the License.

//! # Generalized Message Queue Pallet

//!

//! Provides generalized message queuing and processing capabilities on a per-queue basis for

//! arbitrary use-cases.

//!

//! # Design Goals

//!

//! 1. Minimal assumptions about `Message`s and `MessageOrigin`s. Both should be MEL bounded blobs.

//!  This ensures the generality and reusability of the pallet.

//! 2. Well known and tightly limited pre-dispatch PoV weights, especially for message execution.

//!  This is paramount for the success of the pallet since message execution is done in

//!  `on_initialize` which must _never_ under-estimate its PoV weight. It also needs a frugal PoV

//!  footprint since PoV is scarce and this is (possibly) done in every block. This must also hold

//! in  the presence of unpredictable message size distributions.

//! 3. Usable as XCMP, DMP and UMP message/dispatch queue - possibly through adapter types.

//!

//! # Design

//!

//! The pallet has means to enqueue, store and process messages. This is implemented by having

//! *queues* which store enqueued messages and can be *served* to process said messages. A queue is

//! identified by its origin in the `BookStateFor`. Each message has an origin which defines into

//! which queue it will be stored. Messages are stored by being appended to the last [`Page`] of a

//! book. Each book keeps track of its pages by indexing `Pages`. The `ReadyRing` contains all

//! queues which hold at least one unprocessed message and are thereby *ready* to be serviced. The

//! `ServiceHead` indicates which *ready* queue is the next to be serviced.

//! The pallet implements [`frame_support::traits::EnqueueMessage`],

//! [`frame_support::traits::ServiceQueues`] and has [`frame_support::traits::ProcessMessage`] and

//! [`OnQueueChanged`] hooks to communicate with the outside world.

//!

//! NOTE: The storage items are not linked since they are not public.

//!

//! **Message Execution**

//!

//! Executing a message is offloaded to the [`Config::MessageProcessor`] which contains the actual

//! logic of how to handle the message since they are blobs. Storage changes are not rolled back on

//! error.

//!

//! A failed message can be temporarily or permanently overweight. The pallet will perpetually try

//! to execute a temporarily overweight message. A permanently overweight message is skipped and

//! must be executed manually.

//!

//! **Reentrancy**

//!

//! This pallet has two entry points for executing (possibly recursive) logic;

//! [`Pallet::service_queues`] and [`Pallet::execute_overweight`]. Both entry points are guarded by

//! the same mutex to error on reentrancy. The only functions that are explicitly **allowed** to be

//! called by a message processor are: [`Pallet::enqueue_message`] and

//! [`Pallet::enqueue_messages`]. All other functions are forbidden and error with

//! [`Error::RecursiveDisallowed`].

//!

//! **Pagination**

//!

//! Queues are stored in a *paged* manner by splitting their messages into [`Page`]s. This results

//! in a lot of complexity when implementing the pallet but is completely necessary to achieve the

//! second #[Design Goal](design-goals). The problem comes from the fact a message can *possibly* be

//! quite large, lets say 64KiB. This then results in a *MEL* of at least 64KiB which results in a

//! PoV of at least 64KiB. Now we have the assumption that most messages are much shorter than their

//! maximum allowed length. This would result in most messages having a pre-dispatch PoV size which

//! is much larger than their post-dispatch PoV size, possibly by a factor of thousand. Disregarding

//! this observation would cripple the processing power of the pallet since it cannot straighten out

//! this discrepancy at runtime. Conceptually, the implementation is packing as many messages into a

//! single bounded vec, as actually fit into the bounds. This reduces the wasted PoV.

//!

//! **Page Data Layout**

//!

//! A Page contains a heap which holds all its messages. The heap is built by concatenating

//! `(ItemHeader, Message)` pairs. The [`ItemHeader`] contains the length of the message which is

//! needed for retrieving it. This layout allows for constant access time of the next message and

//! linear access time for any message in the page. The header must remain minimal to reduce its PoV

//! impact.

//!

//! **Weight Metering**

//!

//! The pallet utilizes the [`sp_weights::WeightMeter`] to manually track its consumption to always

//! stay within the required limit. This implies that the message processor hook can calculate the

//! weight of a message without executing it. This restricts the possible use-cases but is necessary

//! since the pallet runs in `on_initialize` which has a hard weight limit. The weight meter is used

//! in a way that `can_accrue` and `check_accrue` are always used to check the remaining weight of

//! an operation before committing to it. The process of exiting due to insufficient weight is

//! termed "bailing".

//!

//! # Scenario: Message enqueuing

//!

//! A message `m` is enqueued for origin `o` into queue `Q[o]` through

//! [`frame_support::traits::EnqueueMessage::enqueue_message`]`(m, o)`.

//!

//! First the queue is either loaded if it exists or otherwise created with empty default values.

//! The message is then inserted to the queue by appended it into its last `Page` or by creating a

//! new `Page` just for `m` if it does not fit in there. The number of messages in the `Book` is

//! incremented.

//!

//! `Q[o]` is now *ready* which will eventually result in `m` being processed.

//!

//! # Scenario: Message processing

//!

//! The pallet runs each block in `on_initialize` or when being manually called through

//! [`frame_support::traits::ServiceQueues::service_queues`].

//!

//! First it tries to "rotate" the `ReadyRing` by one through advancing the `ServiceHead` to the

//! next *ready* queue. It then starts to service this queue by servicing as many pages of it as

//! possible. Servicing a page means to execute as many message of it as possible. Each executed

//! message is marked as *processed* if the [`Config::MessageProcessor`] return Ok. An event

//! [`Event::Processed`] is emitted afterwards. It is possible that the weight limit of the pallet

//! will never allow a specific message to be executed. In this case it remains as unprocessed and

//! is skipped. This process stops if either there are no more messages in the queue or the

//! remaining weight became insufficient to service this queue. If there is enough weight it tries

//! to advance to the next *ready* queue and service it. This continues until there are no more

//! queues on which it can make progress or not enough weight to check that.

//!

//! # Scenario: Overweight execution

//!

//! A permanently over-weight message which was skipped by the message processing will never be

//! executed automatically through `on_initialize` nor by calling

//! [`frame_support::traits::ServiceQueues::service_queues`].

//!

//! Manual intervention in the form of

//! [`frame_support::traits::ServiceQueues::execute_overweight`] is necessary. Overweight messages

//! emit an [`Event::OverweightEnqueued`] event which can be used to extract the arguments for

//! manual execution. This only works on permanently overweight messages. There is no guarantee that

//! this will work since the message could be part of a stale page and be reaped before execution

//! commences.

//!

//! # Terminology

//!

//! - `Message`: A blob of data into which the pallet has no introspection, defined as

//! [`BoundedSlice<u8, MaxMessageLenOf<T>>`]. The message length is limited by [`MaxMessageLenOf`]

//! which is calculated from [`Config::HeapSize`] and [`ItemHeader::max_encoded_len()`].

//! - `MessageOrigin`: A generic *origin* of a message, defined as [`MessageOriginOf`]. The

//! requirements for it are kept minimal to remain as generic as possible. The type is defined in

//! [`frame_support::traits::ProcessMessage::Origin`].

//! - `Page`: An array of `Message`s, see [`Page`]. Can never be empty.

//! - `Book`: A list of `Page`s, see [`BookState`]. Can be empty.

//! - `Queue`: A `Book` together with an `MessageOrigin` which can be part of the `ReadyRing`. Can

//!   be empty.

//! - `ReadyRing`: A double-linked list which contains all *ready* `Queue`s. It chains together the

//!   queues via their `ready_neighbours` fields. A `Queue` is *ready* if it contains at least one

//!   `Message` which can be processed. Can be empty.

//! - `ServiceHead`: A pointer into the `ReadyRing` to the next `Queue` to be serviced.

//! - (`un`)`processed`: A message is marked as *processed* after it was executed by the pallet. A

//!   message which was either: not yet executed or could not be executed remains as `unprocessed`

//!   which is the default state for a message after being enqueued.

//! - `knitting`/`unknitting`: The means of adding or removing a `Queue` from the `ReadyRing`.

//! - `MEL`: The Max Encoded Length of a type, see [`codec::MaxEncodedLen`].

//! - `Reentrance`: To enter an execution context again before it has completed.

//!

//! # Properties

//!

//! **Liveness - Enqueueing**

//!

//! It is always possible to enqueue any message for any `MessageOrigin`.

//!

//! **Liveness - Processing**

//!

//! `on_initialize` always respects its finite weight-limit.

//!

//! **Progress - Enqueueing**

//!

//! An enqueued message immediately becomes *unprocessed* and thereby eligible for execution.

//!

//! **Progress - Processing**

//!

//! The pallet will execute at least one unprocessed message per block, if there is any. Ensuring

//! this property needs careful consideration of the concrete weights, since it is possible that the

//! weight limit of `on_initialize` never allows for the execution of even one message; trivially if

//! the limit is set to zero. `integrity_test` can be used to ensure that this property holds.

//!

//! **Fairness - Enqueuing**

//!

//! Enqueueing a message for a specific `MessageOrigin` does not influence the ability to enqueue a

//! message for the same of any other `MessageOrigin`; guaranteed by **Liveness - Enqueueing**.

//!

//! **Fairness - Processing**

//!

//! The average amount of weight available for message processing is the same for each queue if the

//! number of queues is constant. Creating a new queue must therefore be, possibly economically,

//! expensive. Currently this is archived by having one queue per para-chain/thread, which keeps the

//! number of queues within `O(n)` and should be "good enough".

#![deny(missing_docs)]

#![cfg_attr(not(feature = "std"), no_std)]

mod benchmarking;

mod integration_test;

mod mock;

pub mod mock_helpers;

mod tests;

pub mod weights;

extern crate alloc;

use alloc::{vec, vec::Vec};

use codec::{Codec, Decode, Encode, MaxEncodedLen};

use core::{fmt::Debug, ops::Deref};

use frame_support::{

	defensive,

	pallet_prelude::*,

	traits::{

		Defensive, DefensiveSaturating, DefensiveTruncateFrom, EnqueueMessage,

		ExecuteOverweightError, Footprint, ProcessMessage, ProcessMessageError, QueueFootprint,

		QueuePausedQuery, ServiceQueues,

	},

	BoundedSlice, CloneNoBound, DefaultNoBound,

};

use frame_system::pallet_prelude::*;

pub use pallet::*;

use scale_info::TypeInfo;

use sp_arithmetic::traits::{BaseArithmetic, Unsigned};

use sp_core::{defer, H256};

use sp_runtime::{

	traits::{One, Zero},

	SaturatedConversion, Saturating,

};

use sp_weights::WeightMeter;

pub use weights::WeightInfo;

/// Type for identifying a page.

type PageIndex = u32;

/// Data encoded and prefixed to the encoded `MessageItem`.

pub struct ItemHeader<Size> {

	/// The length of this item, not including the size of this header. The next item of the page

	/// follows immediately after the payload of this item.

	payload_len: Size,

	/// Whether this item has been processed.

	is_processed: bool,

}

/// A page of messages. Pages always contain at least one item.

#[derive(

)]

#[scale_info(skip_type_params(HeapSize))]

#[codec(mel_bound(Size: MaxEncodedLen))]

pub struct Page<Size: Into<u32> + Debug + Clone + Default, HeapSize: Get<Size>> {

	/// Messages remaining to be processed; this includes overweight messages which have been

	/// skipped.

	remaining: Size,

	/// The size of all remaining messages to be processed.

	///

	/// Includes overweight messages outside of the `first` to `last` window.

	remaining_size: Size,

	/// The number of items before the `first` item in this page.

	first_index: Size,

	/// The heap-offset of the header of the first message item in this page which is ready for

	/// processing.

	first: Size,

	/// The heap-offset of the header of the last message item in this page.

	last: Size,

	/// The heap. If `self.offset == self.heap.len()` then the page is empty and should be deleted.

	heap: BoundedVec<u8, IntoU32<HeapSize, Size>>,

}

impl<

		Size: BaseArithmetic + Unsigned + Copy + Into<u32> + Codec + MaxEncodedLen + Debug + Default,

		HeapSize: Get<Size>,

	> Page<Size, HeapSize>

{

	/// Create a [`Page`] from one unprocessed message.

	/// Try to append one message to a page.

			// Can't fit.

	/// Returns the first message in the page without removing it.

	///

	/// SAFETY: Does not panic even on corrupted storage.

				// impossible to truncate since is sliced up from `self.heap: BoundedVec<u8,

				// HeapSize>`

	/// Point `first` at the next message, marking the first as processed if `is_processed` is true.

	/// Return the message with index `index` in the form of `(position, processed, message)`.

		}

	/// Set the `is_processed` flag for the item at `pos` to be `true` if not already and decrement

	/// the `remaining` counter of the page.

	///

	/// Does nothing if no [`ItemHeader`] could be decoded at the given position.

	/// Returns whether the page is *complete* which means that no messages remain.

}

/// A single link in the double-linked Ready Ring list.

pub struct Neighbours<MessageOrigin> {

	/// The previous queue.

	prev: MessageOrigin,

	/// The next queue.

	next: MessageOrigin,

}

/// The state of a queue as represented by a book of its pages.

///

/// Each queue has exactly one book which holds all of its pages. All pages of a book combined

/// contain all of the messages of its queue; hence the name *Book*.

/// Books can be chained together in a double-linked fashion through their `ready_neighbours` field.

pub struct BookState<MessageOrigin> {

	/// The first page with some items to be processed in it. If this is `>= end`, then there are

	/// no pages with items to be processing in them.

	begin: PageIndex,

	/// One more than the last page with some items to be processed in it.

	end: PageIndex,

	/// The number of pages stored at present.

	///

	/// This might be larger than `end-begin`, because we keep pages with unprocessed overweight

	/// messages outside of the end/begin window.

	count: PageIndex,

	/// If this book has any ready pages, then this will be `Some` with the previous and next

	/// neighbours. This wraps around.

	ready_neighbours: Option<Neighbours<MessageOrigin>>,

	/// The number of unprocessed messages stored at present.

	message_count: u64,

	/// The total size of all unprocessed messages stored at present.

	size: u64,

}

impl<MessageOrigin> Default for BookState<MessageOrigin> {

}

impl<MessageOrigin> From<BookState<MessageOrigin>> for QueueFootprint {

}

/// Handler code for when the items in a queue change.

pub trait OnQueueChanged<Id> {

	/// Note that the queue `id` now has `item_count` items in it, taking up `items_size` bytes.

	fn on_queue_changed(id: Id, fp: QueueFootprint);

}

impl<Id> OnQueueChanged<Id> for () {

}

pub mod pallet {

	use super::*;

	pub struct Pallet<T>(_);

	/// The module configuration trait.

	#[pallet::config]

	pub trait Config: frame_system::Config {

		/// The overarching event type.

		type RuntimeEvent: From<Event<Self>> + IsType<<Self as frame_system::Config>::RuntimeEvent>;

		/// Weight information for extrinsics in this pallet.

		type WeightInfo: WeightInfo;

		/// Processor for a message.

		///

		/// Storage changes are not rolled back on error.

		///

		/// # Benchmarking

		///

		/// Must be set to [`mock_helpers::NoopMessageProcessor`] for benchmarking.

		/// Other message processors that consumes exactly (1, 1) weight for any give message will

		/// work as well. Otherwise the benchmarking will also measure the weight of the message

		/// processor, which is not desired.

		type MessageProcessor: ProcessMessage;

		/// Page/heap size type.

		type Size: BaseArithmetic

			+ Unsigned

			+ Copy

			+ Into<u32>

			+ Member

			+ Encode

			+ Decode

			+ MaxEncodedLen

			+ TypeInfo

			+ Default;

		/// Code to be called when a message queue changes - either with items introduced or

		/// removed.

		type QueueChangeHandler: OnQueueChanged<<Self::MessageProcessor as ProcessMessage>::Origin>;

		/// Queried by the pallet to check whether a queue can be serviced.

		///

		/// This also applies to manual servicing via `execute_overweight` and `service_queues`. The

		/// value of this is only polled once before servicing the queue. This means that changes to

		/// it that happen *within* the servicing will not be reflected.

		type QueuePausedQuery: QueuePausedQuery<<Self::MessageProcessor as ProcessMessage>::Origin>;

		/// The size of the page; this implies the maximum message size which can be sent.

		///

		/// A good value depends on the expected message sizes, their weights, the weight that is

		/// available for processing them and the maximal needed message size. The maximal message

		/// size is slightly lower than this as defined by [`MaxMessageLenOf`].

		#[pallet::constant]

		type HeapSize: Get<Self::Size>;

		/// The maximum number of stale pages (i.e. of overweight messages) allowed before culling

		/// can happen. Once there are more stale pages than this, then historical pages may be

		/// dropped, even if they contain unprocessed overweight messages.

		#[pallet::constant]

		type MaxStale: Get<u32>;

		/// The amount of weight (if any) which should be provided to the message queue for

		/// servicing enqueued items `on_initialize`.

		///

		/// This may be legitimately `None` in the case that you will call

		/// `ServiceQueues::service_queues` manually or set [`Self::IdleMaxServiceWeight`] to have

		/// it run in `on_idle`.

		#[pallet::constant]

		type ServiceWeight: Get<Option<Weight>>;

		/// The maximum amount of weight (if any) to be used from remaining weight `on_idle` which

		/// should be provided to the message queue for servicing enqueued items `on_idle`.

		/// Useful for parachains to process messages at the same block they are received.

		///

		/// If `None`, it will not call `ServiceQueues::service_queues` in `on_idle`.

		#[pallet::constant]

		type IdleMaxServiceWeight: Get<Option<Weight>>;

	}

	pub enum Event<T: Config> {

		ProcessingFailed {

		Processed {

		OverweightEnqueued {

		PageReaped {

	}

	pub enum Error<T> {

	}

	/// The index of the first and last (non-empty) pages.

	pub(super) type BookStateFor<T: Config> =

		StorageMap<_, Twox64Concat, MessageOriginOf<T>, BookState<MessageOriginOf<T>>, ValueQuery>;

	/// The origin at which we should begin servicing.

	pub(super) type ServiceHead<T: Config> = StorageValue<_, MessageOriginOf<T>, OptionQuery>;

	/// The map of page indices to pages.

	pub(super) type Pages<T: Config> = StorageDoubleMap<

		_,

		Twox64Concat,

		MessageOriginOf<T>,

		Twox64Concat,

		PageIndex,

		Page<T::Size, T::HeapSize>,

		OptionQuery,

	>;

	impl<T: Config> Hooks<BlockNumberFor<T>> for Pallet<T> {

			} else {

			}

				// Make use of the remaining weight to process enqueued messages.

			} else {

			}

		#[cfg(feature = "try-runtime")]

		/// Check all compile-time assumptions about [`crate::Config`].

		#[cfg(test)]

		fn integrity_test() {

			Self::do_integrity_test().expect("Pallet config is valid; qed")

		}

	}

	impl<T: Config> Pallet<T> {

		/// Remove a page which has no more messages remaining to be processed or is stale.

		#[pallet::call_index(0)]

		#[pallet::weight(T::WeightInfo::reap_page())]

		pub fn reap_page(

			origin: OriginFor<T>,

			message_origin: MessageOriginOf<T>,

			page_index: PageIndex,

		}

		/// Execute an overweight message.

		///

		/// Temporary processing errors will be propagated whereas permanent errors are treated

		/// as success condition.

		///

		/// - `origin`: Must be `Signed`.

		/// - `message_origin`: The origin from which the message to be executed arrived.

		/// - `page`: The page in the queue in which the message to be executed is sitting.

		/// - `index`: The index into the queue of the message to be executed.

		/// - `weight_limit`: The maximum amount of weight allowed to be consumed in the execution

		///   of the message.

		///

		/// Benchmark complexity considerations: O(index + weight_limit).

		#[pallet::call_index(1)]

		#[pallet::weight(

			T::WeightInfo::execute_overweight_page_updated().max(

			T::WeightInfo::execute_overweight_page_removed()).saturating_add(*weight_limit)

		)]

		pub fn execute_overweight(

			origin: OriginFor<T>,

			message_origin: MessageOriginOf<T>,

			page: PageIndex,

			index: T::Size,

			weight_limit: Weight,

		}

	}

}

/// The status of a page after trying to execute its next message.

#[derive(PartialEq, Debug)]

enum PageExecutionStatus {

	/// The execution bailed because there was not enough weight remaining.

	Bailed,

	/// The page did not make any progress on its execution.

	///

	/// This is a transient condition and can be handled by retrying - exactly like [Bailed].

	NoProgress,

	/// No more messages could be loaded. This does _not_ imply `page.is_complete()`.

	///

	/// The reasons for this status are:

	///  - The end of the page is reached but there could still be skipped messages.

	///  - The storage is corrupted.

	NoMore,

}

/// The status after trying to execute the next item of a [`Page`].

#[derive(PartialEq, Debug)]

enum ItemExecutionStatus {

	/// The execution bailed because there was not enough weight remaining.

	Bailed,

	/// The item did not make any progress on its execution.

	///

	/// This is a transient condition and can be handled by retrying - exactly like [Bailed].

	NoProgress,

	/// The item was not found.

	NoItem,

	/// Whether the execution of an item resulted in it being processed.

	///

	/// One reason for `false` would be permanently overweight.

	Executed(bool),

}

/// The status of an attempt to process a message.

#[derive(PartialEq)]

enum MessageExecutionStatus {

	/// There is not enough weight remaining at present.

	InsufficientWeight,

	/// There will never be enough weight.

	Overweight,

	/// The message was processed successfully.

	Processed,

	/// The message was processed and resulted in a, possibly permanent, error.

	Unprocessable { permanent: bool },

	/// The stack depth limit was reached.

	///

	/// We cannot just return `Unprocessable` in this case, because the processability of the

	/// message depends on how the function was called. This may be a permanent error if it was

	/// called by a top-level function, or a transient error if it was already called in a nested

	/// function.

	StackLimitReached,

}

impl<T: Config> Pallet<T> {

	/// Knit `origin` into the ready ring right at the end.

	///

	/// Return the two ready ring neighbours of `origin`.

		} else {

		}

			);

			// Service queue empty.

		} else {

			} else {

			}

		}

	/// Tries to bump the current `ServiceHead` to the next ready queue.

	///

	/// Returns the current head if it got be bumped and `None` otherwise.

			} else {

			}

		} else {

		}

	/// The maximal weight that a single message can consume.

	///

	/// Any message using more than this will be marked as permanently overweight and not

	/// automatically re-attempted. Returns `None` if the servicing of a message cannot begin.

	/// `Some(0)` means that only messages with no weight may be served.

	/// The overhead of servicing a single message.

	/// Checks invariants of the pallet config.

	///

	/// The results of this can only be relied upon if the config values are set to constants.

	#[cfg(test)]

	fn do_integrity_test() -> Result<(), String> {

		ensure!(!MaxMessageLenOf::<T>::get().is_zero(), "HeapSize too low");

		if let Some(service) = T::ServiceWeight::get() {

			if Self::max_message_weight(service).is_none() {

				return Err(format!(

					"ServiceWeight too low: {}. Must be at least {}",

					service,

					Self::single_msg_overhead(),

				))

			}

		}

		Ok(())

	}

			// Already have a page in progress - attempt to append.

				None => {

				},

			};

		} else {

			);

			// insert into ready queue.

				Err(()) => {

				},

			}

		}

		// No room on the page or no page - link in a new page.

	/// Try to execute a single message that was marked as overweight.

	///

	/// The `weight_limit` is the weight that can be consumed to execute the message. The base

	/// weight of the function it self must be measured by the caller.

		}

	/// Same as `do_execute_overweight` but must be called while holding the `service_mutex`.

		);

		use MessageExecutionStatus::*;

			StackLimitReached | Unprocessable { permanent: false } =>

			Unprocessable { permanent: true } | Processed => {

					);

				// no need to consider .first or ready ring since processing an overweight page

				// would not alter that state.

				} else {

				};

			},

		}

	/// Remove a stale page or one which has no more messages remaining to be processed.

		}

	/// Same as `do_reap_page` but must be called while holding the `service_mutex`.

		// definitely reapable if the page has no messages in it.

			// The amount beyond the maximum which are being used. If it's not beyond the maximum

			// then we exit now since no culling is needed.

			};

			// The special formula which tells us how deep into index-history we will pages. As

			// the overflow is greater (and thus the need to drop items from storage is more urgent)

			// this is reduced, allowing a greater range of pages to be culled.

			// With a minimum `overflow` (`1`), this returns `max_stale ** 2`, indicating we only

			// cull beyond that number of indices deep into history.

			// At this overflow increases, our depth reduces down to a limit of `max_stale`. We

			// never want to reduce below this since this will certainly allow enough pages to be

			// culled in order to bring `overflow` back to zero.

	/// Execute any messages remaining to be processed in the queue of `origin`, using up to

	/// `weight_limit` to do so. Any messages which would take more than `overweight_limit` to

	/// execute are deemed overweight and ignored.

		use PageExecutionStatus::*;

		{

				// Store the page progress and do not go to the next one.

				// Go to the next page if this one is at the end.

		}

			// No longer ready - unknit.

	/// Service as many messages of a page as possible.

	///

	/// Returns how many messages were processed and the page's status.

		use PageExecutionStatus::*;

		{

			None => {

			},

		};

		// Execute as many messages as possible.

				// Keep going as long as we make progress...

			}

		};

	/// Execute the next message of a page.

		use MessageExecutionStatus::*;

		// This ugly pre-checking is needed for the invariant

		// "we never bail if a page became complete".

		};

		};

	/// Ensure the correctness of state of this pallet.

	///

	/// # Assumptions-

	///

	/// If `serviceHead` points to a ready Queue, then BookState of that Queue has:

	///

	/// * `message_count` > 0

	/// * `size` > 0

	/// * `end` > `begin`

	/// * Some(ready_neighbours)

	/// * If `ready_neighbours.next` == self.origin, then `ready_neighbours.prev` == self.origin

	///   (only queue in ring)

	///

	/// For Pages(begin to end-1) in BookState:

	///

	/// * `remaining` > 0

	/// * `remaining_size` > 0

	/// * `first` <= `last`

	/// * Every page can be decoded into peek_* functions

	#[cfg(any(test, feature = "try-runtime", feature = "std"))]

		);

		// Checking memory corruption for Pages

		);

		// Basic checks for each book

		}

		//loop around this origin

			);