corevm-engine 0.1.28

use crate::{AudioModeExt as _, InputLimitReached, InputStats, VideoModeExt as _};
use alloc::{collections::VecDeque, vec, vec::Vec};
use bytes::Bytes;
use codec::{Compact, Encode};
use corevm_codec::video;
use corevm_host::{
	AudioMode, CoreVmExtrinsics, PageNum, PageSegmentOps, ServiceMessage, VideoMode,
};
use jam_types::{max_input, SegmentBytes, VecMap, VecSet, SEGMENT_LEN};

mod chunked;
pub(crate) use self::chunked::*;

const SIGNATURE_LEN: usize = 64;

/// Internal bounded input state.
enum State {
	/// Input size limit hasn't been reached.
	///
	/// The parameter is the number of input chunks read so far.
	Reading(u32),
	/// Input size limit has been reached.
	///
	/// The parameter is the number of input chunks read so far.
	LimitReached(u32),
}

/// Work package input (package + extrinsics + imports), size of which is limited by [`max_input`].
///
/// Base input size includes only the size of the work package. Each operation that reads a
/// piece of input from the extrinsics increases the input size by the amount of bytes that were
/// read. When this amount reaches [`max_input`] the operation fails with [`InputLimitReached`] and
/// the VM is suspended.
///
/// This input is only useful within the context of CoreVM builder to determine how many pages,
/// messages and what portions of input streams to include in the extrinsics. Within the context of
/// JAM node extrinsics should only include what the guest was able to read during the simulation.
pub struct BoundedInput {
	/// Includes work package size and the size of all components of the extrinsics, excludes
	/// lengths.
	///
	/// Use [`Self::input_size`] to compute actual input size.
	base_input_size: u32,
	max_input_size: u32,

	imported_memory_pages: VecMap<PageNum, SegmentBytes>,
	accessed_memory_pages: VecSet<PageNum>,

	service_messages: VecDeque<(ServiceMessage, Bytes)>,
	num_service_messages_read: u32,
	/// Equals `true` if the current inter-service message has already been peeked, but not yet
	/// popped from the queue.
	///
	/// Peeking the message for the first time increases the input size.
	service_message_peeked: bool,

	host_messages: ChunkedVecDequeBytes,

	console: ChunkedBytesInput,

	video: ChunkedBytesInput,
	/// Video input decoder.
	video_decoder: Option<video::Decoder>,
	num_video_frames_read: u32,

	audio: ChunkedBytesInput,

	/// Item `i` of this vector is the sum of input chunk lengths that precede input chunk `i`.
	///
	/// The first element is always zero. The last element (its index equals the total number of
	/// chunks) equals the total length of input chunks.
	cumulative_input_chunk_size: Vec<u32>,
	/// The number of input chunks that were read excluding the chunk that caused input size to
	/// overflow (if any).
	state: State,
}

impl BoundedInput {
	pub fn new(work_package_encoded_size: u32, extrinsics: CoreVmExtrinsics) -> Self {
		let mut imported_memory_pages =
			VecMap::with_capacity(extrinsics.imported_memory_pages.len());
		for segment in extrinsics.imported_memory_pages.into_iter() {
			let page_number = segment.as_array().page_number();
			// It's programming error to provide duplicate page numbers via extrinsics.
			assert!(
				imported_memory_pages.insert(page_number, segment).is_none(),
				"Duplicate memory page number: {page_number}"
			);
		}
		let mut console = VecDeque::new();
		let mut video = VecDeque::new();
		let mut audio = VecDeque::new();
		let mut host_messages = VecDeque::new();
		let mut cumulative_input_chunk_size = Vec::with_capacity(extrinsics.input_chunks.len() + 1);
		let mut sum = 0_u32;
		for chunk in extrinsics.input_chunks.into_iter() {
			let chunk_size = chunk.encoded_size() + SIGNATURE_LEN;
			console.push_back(chunk.console);
			video.push_back(chunk.video);
			audio.push_back(chunk.audio);
			host_messages.push_back(VecDeque::from(chunk.host_messages));
			cumulative_input_chunk_size.push(sum);
			sum += chunk_size as u32;
		}
		cumulative_input_chunk_size.push(sum);
		debug_assert_eq!(0, cumulative_input_chunk_size[0]);
		Self {
			base_input_size: work_package_encoded_size,
			max_input_size: max_input(),
			imported_memory_pages,
			accessed_memory_pages: VecSet::new(),
			service_messages: extrinsics.incoming_service_messages,
			num_service_messages_read: 0,
			service_message_peeked: false,
			host_messages: ChunkedVecDequeBytes::new(host_messages),
			console: ChunkedBytesInput::new(console),
			video: ChunkedBytesInput::new(video),
			num_video_frames_read: 0,
			video_decoder: None,
			audio: ChunkedBytesInput::new(audio),
			cumulative_input_chunk_size,
			state: State::Reading(0),
		}
	}

	/// Returns statistics that includes accessed imported memory pages, what portions of the input
	/// streams were read and how many messages were processed.
	pub fn into_stats(self) -> InputStats {
		InputStats {
			accessed_imported_pages: self.accessed_memory_pages,
			num_service_messages_read: self.num_service_messages_read,
			num_input_chunks_read: match self.state {
				State::Reading(n) => n,
				State::LimitReached(n) => n,
			},
		}
	}

	fn input_size(&self) -> u32 {
		let input_chunks_size =
			self.cumulative_input_chunk_size[self.num_input_chunks_read() as usize];
		self.base_input_size +
			Compact(self.num_service_messages_read).encoded_size() as u32 +
			input_chunks_size
	}

	/// Imports specified memory page from the input and returns the corresponding segment.
	///
	/// Returns an error if importing a page increases the size of the input beyond [`max_input`].
	pub fn import_page(
		&mut self,
		page: PageNum,
	) -> Result<Option<SegmentBytes>, InputLimitReached> {
		self.check_limit_reached()?;
		let Some(segment) = self.imported_memory_pages.get(&page) else {
			return Ok(None);
		};
		if !self.accessed_memory_pages.insert(page) {
			// The page has already been checked and counted.
			return Ok(Some(segment.clone()));
		}
		self.base_input_size += SEGMENT_LEN as u32;
		if self.input_size() > self.max_input_size {
			self.base_input_size -= SEGMENT_LEN as u32;
			self.accessed_memory_pages.remove(&page);
			self.set_limit_reached();
			return Err(InputLimitReached);
		}
		Ok(Some(segment.clone()))
	}

	/// Peeks next inter-service message from the input.
	///
	/// Returns an error if this increases the size of the input beyond [`max_input`].
	pub fn peek_service_message(
		&mut self,
	) -> Result<Option<(ServiceMessage, Bytes)>, InputLimitReached> {
		self.check_limit_reached()?;
		match self.service_messages.front() {
			Some(element) => {
				if !self.service_message_peeked {
					let encoded_size = element.encoded_size() as u32;
					self.service_message_peeked = true;
					self.num_service_messages_read += 1;
					self.base_input_size += encoded_size;
					if self.input_size() > self.max_input_size {
						self.service_message_peeked = false;
						self.num_service_messages_read -= 1;
						self.base_input_size -= encoded_size;
						self.set_limit_reached();
						return Err(InputLimitReached);
					}
				}
				Ok(Some(element.clone()))
			},
			None => Ok(None),
		}
	}

	/// Extracts next inter-service message from the input.
	///
	/// Returns an error if this increases the size of the input beyond [`max_input`].
	pub fn pop_service_message(
		&mut self,
	) -> Result<Option<(ServiceMessage, Bytes)>, InputLimitReached> {
		self.check_limit_reached()?;
		match self.service_messages.pop_front() {
			Some(element) => {
				if !self.service_message_peeked {
					let encoded_size = element.encoded_size() as u32;
					self.num_service_messages_read += 1;
					self.base_input_size += encoded_size;
					if self.input_size() > self.max_input_size {
						self.service_messages.push_front(element);
						self.num_service_messages_read -= 1;
						self.base_input_size -= encoded_size;
						self.set_limit_reached();
						return Err(InputLimitReached);
					}
				}
				self.service_message_peeked = false;
				Ok(Some(element))
			},
			None => Ok(None),
		}
	}

	/// Peeks next host message in the input.
	///
	/// Returns an error if this increases the size of the input beyond [`max_input`].
	pub fn peek_host_message(&mut self) -> Result<Option<Bytes>, InputLimitReached> {
		let last_num_chunks_read = self.check_limit_reached()?;
		let message = self.host_messages.peek_front_skipping_empty_chunks();
		self.check_input_size(last_num_chunks_read)?;
		Ok(message)
	}

	/// Extracts next host message from the input.
	///
	/// Returns an error if this increases the size of the input beyond [`max_input`].
	pub fn pop_host_message(&mut self) -> Result<Option<Bytes>, InputLimitReached> {
		let last_num_chunks_read = self.check_limit_reached()?;
		let message = self.host_messages.pop_front_skipping_empty_chunks();
		self.check_input_size(last_num_chunks_read)?;
		Ok(message)
	}

	/// Returns `true` if the console input stream is empty.
	///
	/// Returns an error if this increases the size of the input beyond [`max_input`].
	pub fn console_is_empty(&mut self) -> Result<bool, InputLimitReached> {
		let last_num_chunks_read = self.check_limit_reached()?;
		let is_empty = match self.console.peek_within_chunk_skipping_empty_chunks() {
			Some(slice) => slice.is_empty(),
			None => true,
		};
		self.check_input_size(last_num_chunks_read)?;
		Ok(is_empty)
	}

	/// Reads the specified number of bytes from the current chunk of the console input stream and
	/// returns them as bytes.
	///
	/// Might return less bytes than specified.
	///
	/// Returns an error if this increases the size of the input beyond [`max_input`].
	pub fn read_console(&mut self, length: u32) -> Result<Option<Bytes>, InputLimitReached> {
		if length == 0 {
			return Ok(None);
		}
		let last_num_chunks_read = self.check_limit_reached()?;
		let buf = self.console.read_within_chunk_skipping_empty_chunks(length as usize);
		self.check_input_size(last_num_chunks_read)?;
		Ok(buf)
	}

	/// Returns `true` if the video input stream is empty.
	///
	/// Returns an error if this increases the size of the input beyond [`max_input`].
	pub fn video_is_empty(&mut self, mode: VideoMode) -> Result<bool, InputLimitReached> {
		if u64::from(self.num_video_frames_read) >= mode.frames_per_slot() {
			return Ok(true);
		}
		let last_num_chunks_read = self.check_limit_reached()?;
		let is_empty = match self.video.peek_within_chunk_skipping_empty_chunks() {
			Some(slice) => slice.is_empty(),
			None => true,
		};
		self.check_input_size(last_num_chunks_read)?;
		Ok(is_empty)
	}

	/// Reads next video frame from the video stream and returns it as a vector of bytes.
	///
	/// Returns
	/// - `Eof` if the video stream is empty or time limit is reached,
	/// - `InvalidVideoStream` if the video stream is malformed,
	/// - `InputLimitReached` if this increases the size of the input beyond [`max_input`].
	pub fn read_video_frame(&mut self, mode: VideoMode) -> Result<Vec<u8>, ReadVideoFrameError> {
		use ReadVideoFrameError::*;
		if self.video_is_empty(mode)? {
			return Err(Eof);
		}
		let last_num_chunks_read = self.check_limit_reached()?;
		let decoder = match self.video_decoder {
			Some(ref mut decoder) => decoder,
			ref mut decoder @ None => {
				let new_decoder = video::Decoder::new(&mut self.video)?;
				decoder.insert(new_decoder)
			},
		};
		let frame_len = (u32::from(decoder.width().get()) * u32::from(decoder.height().get()))
			.checked_mul(3)
			.ok_or(InvalidVideoStream)?;
		let mut buf = vec![0_u8; frame_len as usize];
		decoder.read_rgb888_frame(&mut self.video, &mut buf)?;
		self.check_input_size(last_num_chunks_read)?;
		self.num_video_frames_read += 1;
		Ok(buf)
	}

	/// Returns `true` if audio input stream is empty or exhausted (i.e. one time-slot worth of data
	/// has been read).
	pub fn audio_is_empty(&mut self, mode: AudioMode) -> Result<bool, InputLimitReached> {
		if self.audio.position() as u64 >= mode.bytes_per_slot() {
			return Ok(true)
		}
		let last_num_chunks_read = self.check_limit_reached()?;
		let is_empty = match self.audio.peek_within_chunk_skipping_empty_chunks() {
			Some(slice) => slice.is_empty(),
			None => true,
		};
		self.check_input_size(last_num_chunks_read)?;
		Ok(is_empty)
	}

	/// Reads up to `length` bytes worth of audio frames from the audio input stream and returns
	/// them as bytes.
	///
	/// This method always returns complete audio frames (a frame is an array of samples, one sample
	/// for each channel): as many as can fit into `length` bytes or as many as are available in
	/// the current input chunk or as many as are allowed by the time-slot limit, whichever number
	/// is smaller.
	///
	/// Returns `None` if the end of the stream has been reached.
	pub fn read_audio_frames(
		&mut self,
		length: u32,
		mode: AudioMode,
	) -> Result<Option<Bytes>, InputLimitReached> {
		let n = u64::from(length)
			.min((mode.bytes_per_slot()).saturating_sub(self.audio.position() as u64))
			as u32;
		if n == 0 {
			return Ok(None);
		}
		let last_num_chunks_read = self.check_limit_reached()?;
		let buf = self.audio.read_within_chunk_skipping_empty_chunks(length as usize);
		self.check_input_size(last_num_chunks_read)?;
		Ok(buf)
	}

	fn num_input_chunks_read(&self) -> u32 {
		self.audio
			.num_chunks_read()
			.max(self.video.num_chunks_read())
			.max(self.console.num_chunks_read())
			.max(self.host_messages.num_chunks_read()) as u32
	}

	/// Ensure that the input limit hasn't been reached in one of the previous method calls.
	///
	/// Returns the number of chunks read so far.
	///
	/// This method should be called _before_ any operation that potentially increases the input
	/// size.
	fn check_limit_reached(&mut self) -> Result<u32, InputLimitReached> {
		match self.state {
			State::Reading(num_chunks_read) => Ok(num_chunks_read),
			State::LimitReached(..) => Err(InputLimitReached),
		}
	}

	/// Finalize reading input chunk.
	///
	/// This method either updates the state with the total number of input chunks read so far
	/// (when the input size is within the limit) or transitions to `State::LimitReached` with the
	/// last known number of input chunks (when the input size limit was reached).
	///
	/// `last_num_chunks_read` is the number of input chunks read prior to reading the last input
	/// chunk.
	///
	/// This method should be called _after_ reading an input chunk.
	fn check_input_size(&mut self, last_num_chunks_read: u32) -> Result<(), InputLimitReached> {
		if self.input_size() > self.max_input_size {
			match self.state {
				State::Reading(..) => {
					// Here we use the last known value extracted from the state as the limit.
					self.state = State::LimitReached(last_num_chunks_read);
				},
				State::LimitReached(..) => unreachable!(
					"check_input_size shouldn't be called when the limit has already been reached"
				),
			}
			return Err(InputLimitReached);
		}
		self.state = State::Reading(self.num_input_chunks_read());
		Ok(())
	}

	/// Transitions the state to `State::LimitReached` if needed.
	///
	/// This method should be called when the input size limit was reached without reading input
	/// chunks.
	fn set_limit_reached(&mut self) {
		match self.state {
			State::Reading(num_chunks_read) => self.state = State::LimitReached(num_chunks_read),
			State::LimitReached(..) => {},
		}
	}
}

#[derive(Debug)]
pub enum ReadVideoFrameError {
	InputLimitReached,
	Eof,
	InvalidVideoStream,
}

impl From<video::InvalidVideoStream> for ReadVideoFrameError {
	fn from(_: video::InvalidVideoStream) -> Self {
		Self::InvalidVideoStream
	}
}

impl From<InputLimitReached> for ReadVideoFrameError {
	fn from(_: InputLimitReached) -> Self {
		Self::InputLimitReached
	}
}

#[cfg(test)]
mod tests {
	use super::*;
	use alloc::vec;
	use core::num::NonZero;
	use corevm_host::{fs, AudioSampleFormat, CoreVmPayload, InputChunk, VmState};
	use jam_types::{max_imports, Authorizer, ExtrinsicSpec, RefineContext, WorkItem, WorkPackage};

	fn create_work_package() -> WorkPackage {
		WorkPackage {
			authorization: Default::default(),
			auth_code_host: 0,
			authorizer: Authorizer { code_hash: Default::default(), config: Default::default() },
			context: RefineContext {
				anchor: Default::default(),
				lookup_anchor: Default::default(),
				lookup_anchor_slot: Default::default(),
				beefy_root: Default::default(),
				state_root: Default::default(),
				prerequisites: Default::default(),
			},
			items: vec![WorkItem {
				service: 0,
				code_hash: Default::default(),
				payload: CoreVmPayload {
					gas: 0,
					vm_state: VmState::initial(),
					exec_ref: fs::BlockRef { service_id: 0, hash: fs::Hash([0; 32]) },
				}
				.encode()
				.into(),
				refine_gas_limit: 0,
				accumulate_gas_limit: 0,
				import_segments: Default::default(),
				extrinsics: vec![
					ExtrinsicSpec { hash: Default::default(), len: 0 };
					CoreVmExtrinsics::MAX_COUNT
				]
				.try_into()
				.unwrap(),
				export_count: 0,
			}]
			.try_into()
			.unwrap(),
		}
	}

	fn calc_input_size(base_input_size: usize, extrinsics: &CoreVmExtrinsics) -> usize {
		base_input_size +
			extrinsics.incoming_service_messages.encoded_size() +
			extrinsics.input_chunks.encoded_size()
	}

	#[test]
	fn import_page_works() {
		let base_encoded_size = create_work_package().encoded_size();
		let mut imported_memory_pages = Vec::new();
		let mut segment = [0_u8; SEGMENT_LEN];
		for i in 0..max_imports() {
			let page = PageNum(1000 + i);
			segment.set_page_number(page);
			imported_memory_pages.push(SegmentBytes::from(&segment));
		}
		let extrinsics = CoreVmExtrinsics { imported_memory_pages, ..Default::default() };
		let mut input = BoundedInput::new(base_encoded_size as u32, extrinsics);
		// Repeat two times to check that we can access the same page multiple times.
		for _ in 0..2 {
			for i in 0..max_imports() {
				let result = input.import_page(PageNum(1000 + i));
				assert!(matches!(result, Ok(Some(..))), "Result {result:?}, page {i}");
			}
		}
	}

	#[test]
	fn pop_service_message_works() {
		let base_encoded_size = create_work_package().encoded_size();
		let num_messages = 10000;
		let incoming_service_messages: VecDeque<_> = (0..num_messages)
			.map(|index| (ServiceMessage { index, source: 0 }, vec![0; 4096].into()))
			.collect();
		assert!(incoming_service_messages.encoded_size() as u32 >= max_input());
		let mut extrinsics = CoreVmExtrinsics { incoming_service_messages, ..Default::default() };
		let mut input = BoundedInput::new(base_encoded_size as u32, extrinsics.clone());
		let mut reached_limit = false;
		for _ in 0..num_messages {
			let result1 = input.peek_service_message();
			let result2 = input.peek_service_message();
			assert_eq!(
				result1.as_ref().map_err(|_| "err"),
				result2.as_ref().map_err(|_| "err"),
				"peek_service_message should be idempotent"
			);
			if result1.is_err() {
				reached_limit = true;
				break;
			}
			input.pop_service_message().unwrap();
		}
		assert!(reached_limit, "Add more service messages");
		let stats = input.into_stats();
		extrinsics
			.incoming_service_messages
			.truncate(stats.num_service_messages_read as usize);
		let input_size = calc_input_size(base_encoded_size, &extrinsics);
		assert!(input_size <= max_input() as usize, "input_size = {input_size}");
	}

	#[test]
	fn pop_host_message_works() {
		let base_encoded_size = create_work_package().encoded_size();
		let mut num_messages = 0_usize;
		let input_chunks: Vec<_> = (0..100)
			.map(|i| {
				num_messages += i;
				InputChunk { host_messages: vec![vec![0; 4096].into(); i], ..Default::default() }
			})
			.collect();
		assert!(input_chunks.encoded_size() as u32 >= max_input());
		let mut extrinsics = CoreVmExtrinsics { input_chunks, ..Default::default() };
		let mut input = BoundedInput::new(base_encoded_size as u32, extrinsics.clone());
		let mut reached_limit = false;
		for _ in 0..num_messages {
			let result1 = input.peek_host_message();
			let result2 = input.peek_host_message();
			assert_eq!(
				result1.as_ref().map_err(|_| "err"),
				result2.as_ref().map_err(|_| "err"),
				"peek_host_message should be idempotent"
			);
			if result1.is_err() {
				reached_limit = true;
				break;
			}
			input.pop_host_message().unwrap();
		}
		assert!(reached_limit, "Add more input chunks");
		assert!(input.input_size() >= max_input());
		let stats = input.into_stats();
		extrinsics.input_chunks.truncate(stats.num_input_chunks_read as usize);
		let input_size = calc_input_size(base_encoded_size, &extrinsics);
		assert!(input_size <= max_input() as usize, "input_size = {input_size}");
	}

	#[test]
	fn read_console_works() {
		let base_encoded_size = create_work_package().encoded_size();
		let num_chunks = 6000;
		let input_chunks: Vec<_> = (0..num_chunks)
			.map(|i| InputChunk { console: vec![0; i].into(), ..Default::default() })
			.collect();
		assert!(input_chunks.encoded_size() as u32 >= max_input());
		let mut extrinsics = CoreVmExtrinsics { input_chunks, ..Default::default() };
		let mut input = BoundedInput::new(base_encoded_size as u32, extrinsics.clone());
		let mut reached_limit = false;
		for length in 0..num_chunks {
			if input.read_console(length as u32).is_err() {
				reached_limit = true;
				break;
			}
		}
		assert!(reached_limit, "Add more input chunks");
		assert!(input.input_size() >= max_input());
		let stats = input.into_stats();
		extrinsics.input_chunks.truncate(stats.num_input_chunks_read as usize);
		let input_size = calc_input_size(base_encoded_size, &extrinsics);
		assert!(input_size <= max_input() as usize, "input_size = {input_size}");
	}

	#[test]
	fn read_video_frame_works() {
		let base_encoded_size = create_work_package().encoded_size();
		let num_video_frames = 400;
		let width = NonZero::new(320).unwrap();
		let height = NonZero::new(200).unwrap();
		let video = {
			let mut output = Vec::new();
			let mut config = video::Config::default();
			config.raw = true;
			let mut encoder = video::Encoder::new(width, height, config);
			encoder.start(&mut output);
			let frame = vec![0_u8; usize::from(width.get()) * usize::from(height.get()) * 3];
			for _ in 0..num_video_frames {
				encoder.write_rgb888_frame(&frame, &mut output);
			}
			encoder.finish(&mut output);
			Bytes::from(output)
		};
		let input_chunks: Vec<_> = video
			.chunks(1000)
			.map(|chunk| InputChunk { video: video.slice_ref(chunk), ..Default::default() })
			.collect();
		assert!(input_chunks.encoded_size() as u32 >= max_input());
		// Use large FPS to trigger OutputLimitReached. With smaller values we would hit time
		// limit first.
		let mode = VideoMode::raw(width.get(), height.get(), u16::MAX).unwrap();
		let mut extrinsics = CoreVmExtrinsics { input_chunks, ..Default::default() };
		let mut input = BoundedInput::new(base_encoded_size as u32, extrinsics.clone());
		let mut reached_limit = false;
		for _ in 0..num_video_frames {
			if input.read_video_frame(mode).is_err() {
				reached_limit = true;
				break;
			}
		}
		assert!(reached_limit, "Add more input chunks");
		assert!(input.input_size() >= max_input());
		let stats = input.into_stats();
		extrinsics.input_chunks.truncate(stats.num_input_chunks_read as usize);
		let input_size = calc_input_size(base_encoded_size, &extrinsics);
		assert!(input_size <= max_input() as usize, "input_size = {input_size}");
	}

	#[test]
	fn read_audio_frames_works() {
		let base_encoded_size = create_work_package().encoded_size();
		let audio_length_in_bytes = 20_000_000;
		let audio = Bytes::from(vec![0_u8; audio_length_in_bytes]);
		// Use large values to trigger OutputLimitReached. With smaller values we would hit time
		// limit first.
		let mode = AudioMode {
			channels: NonZero::new(u8::MAX).unwrap(),
			sample_format: AudioSampleFormat::S16LE,
			sample_rate: NonZero::new(u32::MAX).unwrap(),
		};
		let input_chunks: Vec<_> = audio
			.chunks(1000)
			.map(|chunk| InputChunk { audio: audio.slice_ref(chunk), ..Default::default() })
			.collect();
		assert!(input_chunks.encoded_size() as u32 >= max_input());
		let mut extrinsics = CoreVmExtrinsics { input_chunks, ..Default::default() };
		let mut input = BoundedInput::new(base_encoded_size as u32, extrinsics.clone());
		let mut reached_limit = false;
		let mut iter = 0..audio_length_in_bytes;
		loop {
			let length = iter.by_ref().take(1000).count();
			if length == 0 {
				break;
			}
			if input.read_audio_frames(length as u32, mode).is_err() {
				reached_limit = true;
				break;
			}
		}
		assert!(reached_limit, "Add more input chunks");
		assert!(input.input_size() >= max_input());
		let stats = input.into_stats();
		extrinsics.input_chunks.truncate(stats.num_input_chunks_read as usize);
		let input_size = calc_input_size(base_encoded_size, &extrinsics);
		assert!(input_size <= max_input() as usize, "input_size = {input_size}");
	}
}