Prevent OOM panic in Deserializable::read_batch_from

utils/core/src/errors.rs

@@ -18,6 +18,8 @@ pub enum DeserializationError {
     UnexpectedEOF,
     /// Deserialization has finished but not all bytes have been consumed.
     UnconsumedBytes,
+    /// The requested number of elements could not be allocated.
+    TooManyElements(usize),
     /// An unknown error has occurred.
     UnknownError(String),
 }
@@ -28,6 +30,9 @@ impl fmt::Display for DeserializationError {
             Self::InvalidValue(err_msg) => write!(f, "{err_msg}"),
             Self::UnexpectedEOF => write!(f, "unexpected EOF"),
             Self::UnconsumedBytes => write!(f, "not all bytes were consumed"),
+            Self::TooManyElements(num) => {
+                write!(f, "could not allocate memory for {num} elements")
+            }
             Self::UnknownError(err_msg) => write!(f, "unknown error: {err_msg}"),
         }
     }

utils/core/src/serde/mod.rs

@@ -393,6 +393,34 @@ pub trait Deserializable: Sized {
     }
 }
 
+    /// Reads a sequence of bytes from the provided `source`, attempts to deserialize these bytes
+    /// into a vector with the specified number of `Self` elements, and returns the result.
+    ///
+    /// # Errors
+    /// Returns an error if:
+    /// * The `source` does not contain enough bytes to deserialize the specified number of
+    ///   elements.
+    /// * Bytes read from the `source` do not represent a valid value for `Self` for any of the
+    ///   elements.
+    ///
+    /// Note: if the error occurs, the reader is not rolled back to the state prior to calling
+    /// this function.
+    fn read_batch_from<R: ByteReader>(
+        source: &mut R,
+        num_elements: usize,
+    ) -> Result<Vec<Self>, DeserializationError> {
+        let mut result = Vec::new();
+        result
+            .try_reserve(num_elements)
+            .map_err(|_| DeserializationError::TooManyElements(num_elements))?;
+        for _ in 0..num_elements {
+            let element = Self::read_from(source)?;
+            result.push(element)
+        }
+        Ok(result)
+    }
+}
+
 impl Deserializable for () {
     fn read_from<R: ByteReader>(_source: &mut R) -> Result<Self, DeserializationError> {
         Ok(())