fix(io): make event import iteration fallible

crates/io/msgpack/src/lib.rs

@@ -5,16 +5,23 @@
 //!
 //! File format: sequence of length-prefixed records.
 //! Each record: `[4 bytes: payload length as u32 BE][payload: msgpack-encoded Event<T>]`
-use std::{io::BufReader, marker::PhantomData, path::PathBuf};
+use std::{
+    io::{BufReader, Read},
+    marker::PhantomData,
+    path::PathBuf,
+};
 
 use quent_events::{EntityEvent, Event};
-use quent_io_types::{Exporter, ExporterError, ExporterResult, Importer, ImporterResult};
+use quent_io_types::{
+    Exporter, ExporterError, ExporterResult, Importer, ImporterError, ImporterResult,
+    MAX_FRAME_SIZE_BYTES,
+};
 use serde::{Deserialize, Serialize};
 use tokio::{
     fs::{File, OpenOptions},
     io::{AsyncWriteExt, BufWriter},
 };
-use tracing::{debug, error, warn};
+use tracing::{debug, warn};
 use uuid::Uuid;
 
 /// File extension for MessagePack event files.
@@ -113,6 +120,7 @@ pub struct MsgpackImporterOptions {
 
 pub struct MsgpackImporter<T> {
     reader: BufReader<std::fs::File>,
+    terminated: bool,
     _phantom: PhantomData<T>,
 }
 
@@ -122,6 +130,7 @@ impl<T> MsgpackImporter<T> {
         let file = std::fs::File::open(&path)?;
         Ok(Self {
             reader: BufReader::new(file),
+            terminated: false,
             _phantom: Default::default(),
         })
     }
@@ -133,31 +142,54 @@ impl<T> Iterator for MsgpackImporter<T>
 where
     T: for<'de> Deserialize<'de>,
 {
-    type Item = Event<T>;
+    type Item = ImporterResult<Event<T>>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        use std::io::Read;
+        if self.terminated {
+            return None;
+        }
+
         let mut len_buf = [0u8; 4];
-        match self.reader.read_exact(&mut len_buf) {
-            Ok(()) => {}
-            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return None,
-            Err(e) => {
-                error!("failed to read msgpack length: {e}");
-                return None;
-            }
+        match self.reader.read(&mut len_buf[..1]) {
+            Ok(0) => return None,
+            Ok(_) => {}
+            // The reader position after an I/O failure may not be a frame boundary.
+            Err(error) => return self.fail(error.into()),
+        }
+        if let Err(error) = self.reader.read_exact(&mut len_buf[1..]) {
+            // An incomplete length prefix does not identify the next frame boundary.
+            return self.fail(error.into());
         }
         let len = u32::from_be_bytes(len_buf) as usize;
-        let mut payload = vec![0u8; len];
-        if let Err(e) = self.reader.read_exact(&mut payload) {
-            error!("failed to read msgpack payload: {e}");
-            return None;
+        if len > MAX_FRAME_SIZE_BYTES {
+            // Consuming an unsupported payload could require unbounded I/O before resuming.
+            return self.fail(ImporterError::other(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                format!(
+                    "frame size {len} exceeds the supported maximum of {MAX_FRAME_SIZE_BYTES} bytes"
+                ),
+            )));
+        }
+        let mut payload = Vec::new();
+        if let Err(error) = payload.try_reserve_exact(len) {
+            // Without a payload buffer, this importer cannot decode the current frame.
+            return self.fail(ImporterError::other(error));
+        }
+        payload.resize(len, 0);
+        if let Err(error) = self.reader.read_exact(&mut payload) {
+            // An incomplete payload leaves the reader before the next frame boundary.
+            return self.fail(error.into());
         }
         match rmp_serde::from_slice::<Event<T>>(&payload) {
-            Ok(event) => Some(event),
-            Err(e) => {
-                error!("failed to deserialize msgpack event: {e}");
-                None
-            }
+            Ok(event) => Some(Ok(event)),
+            Err(error) => Some(Err(ImporterError::other(error))),
         }
     }
 }
+
+impl<T> MsgpackImporter<T> {
+    fn fail(&mut self, error: quent_io_types::ImporterError) -> Option<ImporterResult<Event<T>>> {
+        self.terminated = true;
+        Some(Err(error))
+    }
+}

crates/io/ndjson/src/lib.rs

@@ -9,13 +9,15 @@ use std::{
 };
 
 use quent_events::{EntityEvent, Event};
-use quent_io_types::{Exporter, ExporterError, ExporterResult, Importer, ImporterResult};
+use quent_io_types::{
+    Exporter, ExporterError, ExporterResult, Importer, ImporterError, ImporterResult,
+};
 use serde::{Deserialize, Serialize};
 use tokio::{
     fs::{File, OpenOptions},
     io::{AsyncWriteExt, BufWriter},
 };
-use tracing::{debug, error, warn};
+use tracing::{debug, warn};
 use uuid::Uuid;
 
 /// File extension for ndjson event files.
@@ -116,6 +118,7 @@ pub struct NdjsonImporterOptions {
 
 pub struct NdjsonImporter<T> {
     reader: BufReader<std::fs::File>,
+    terminated: bool,
     _phantom: PhantomData<T>,
 }
 
@@ -125,6 +128,7 @@ impl<T> NdjsonImporter<T> {
         let file = std::fs::File::open(&path)?;
         Ok(Self {
             reader: BufReader::new(file),
+            terminated: false,
             _phantom: Default::default(),
         })
     }
@@ -136,25 +140,24 @@ impl<T> Iterator for NdjsonImporter<T>
 where
     T: for<'de> Deserialize<'de>,
 {
-    type Item = Event<T>;
+    type Item = ImporterResult<Event<T>>;
 
     fn next(&mut self) -> Option<Self::Item> {
+        if self.terminated {
+            return None;
+        }
+
         let mut line = String::new();
         match self.reader.read_line(&mut line) {
             Ok(0) => None,
-            Ok(_) => {
-                let trimmed = line.trim_end();
-                match serde_json::from_str::<Event<T>>(trimmed) {
-                    Ok(event) => Some(event),
-                    Err(e) => {
-                        error!("failed to parse ndjson line: {e}");
-                        None
-                    }
-                }
-            }
+            Ok(_) => match serde_json::from_str::<Event<T>>(line.trim_end()) {
+                Ok(event) => Some(Ok(event)),
+                Err(error) => Some(Err(ImporterError::other(error))),
+            },
             Err(e) => {
-                error!("failed to read ndjson: {e}");
-                None
+                // The failed read may have consumed a partial line without its delimiter.
+                self.terminated = true;
+                Some(Err(e.into()))
             }
         }
     }

crates/io/postcard/src/lib.rs

@@ -5,16 +5,23 @@
 //!
 //! File format: sequence of length-prefixed records.
 //! Each record: `[4 bytes: payload length as u32 BE][payload: postcard-encoded Event<T>]`
-use std::{io::BufReader, marker::PhantomData, path::PathBuf};
+use std::{
+    io::{BufReader, Read},
+    marker::PhantomData,
+    path::PathBuf,
+};
 
 use quent_events::{EntityEvent, Event};
-use quent_io_types::{Exporter, ExporterError, ExporterResult, Importer, ImporterResult};
+use quent_io_types::{
+    Exporter, ExporterError, ExporterResult, Importer, ImporterError, ImporterResult,
+    MAX_FRAME_SIZE_BYTES,
+};
 use serde::{Deserialize, Serialize};
 use tokio::{
     fs::{File, OpenOptions},
     io::{AsyncWriteExt, BufWriter},
 };
-use tracing::{debug, error, warn};
+use tracing::{debug, warn};
 use uuid::Uuid;
 
 /// File extension for Postcard event files.
@@ -112,6 +119,7 @@ pub struct PostcardImporterOptions {
 
 pub struct PostcardImporter<T> {
     reader: BufReader<std::fs::File>,
+    terminated: bool,
     _phantom: PhantomData<T>,
 }
 
@@ -121,6 +129,7 @@ impl<T> PostcardImporter<T> {
         let file = std::fs::File::open(&path)?;
         Ok(Self {
             reader: BufReader::new(file),
+            terminated: false,
             _phantom: Default::default(),
         })
     }
@@ -132,31 +141,54 @@ impl<T> Iterator for PostcardImporter<T>
 where
     T: for<'de> Deserialize<'de>,
 {
-    type Item = Event<T>;
+    type Item = ImporterResult<Event<T>>;
 
     fn next(&mut self) -> Option<Self::Item> {
-        use std::io::Read;
+        if self.terminated {
+            return None;
+        }
+
         let mut len_buf = [0u8; 4];
-        match self.reader.read_exact(&mut len_buf) {
-            Ok(()) => {}
-            Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return None,
-            Err(e) => {
-                error!("failed to read postcard length: {e}");
-                return None;
-            }
+        match self.reader.read(&mut len_buf[..1]) {
+            Ok(0) => return None,
+            Ok(_) => {}
+            // The reader position after an I/O failure may not be a frame boundary.
+            Err(error) => return self.fail(error.into()),
+        }
+        if let Err(error) = self.reader.read_exact(&mut len_buf[1..]) {
+            // An incomplete length prefix does not identify the next frame boundary.
+            return self.fail(error.into());
         }
         let len = u32::from_be_bytes(len_buf) as usize;
-        let mut payload = vec![0u8; len];
-        if let Err(e) = self.reader.read_exact(&mut payload) {
-            error!("failed to read postcard payload: {e}");
-            return None;
+        if len > MAX_FRAME_SIZE_BYTES {
+            // Consuming an unsupported payload could require unbounded I/O before resuming.
+            return self.fail(ImporterError::other(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                format!(
+                    "frame size {len} exceeds the supported maximum of {MAX_FRAME_SIZE_BYTES} bytes"
+                ),
+            )));
+        }
+        let mut payload = Vec::new();
+        if let Err(error) = payload.try_reserve_exact(len) {
+            // Without a payload buffer, this importer cannot decode the current frame.
+            return self.fail(ImporterError::other(error));
+        }
+        payload.resize(len, 0);
+        if let Err(error) = self.reader.read_exact(&mut payload) {
+            // An incomplete payload leaves the reader before the next frame boundary.
+            return self.fail(error.into());
         }
         match postcard::from_bytes::<Event<T>>(&payload) {
-            Ok(event) => Some(event),
-            Err(e) => {
-                error!("failed to deserialize postcard event: {e}");
-                None
-            }
+            Ok(event) => Some(Ok(event)),
+            Err(error) => Some(Err(ImporterError::other(error))),
         }
     }
 }
+
+impl<T> PostcardImporter<T> {
+    fn fail(&mut self, error: quent_io_types::ImporterError) -> Option<ImporterResult<Event<T>>> {
+        self.terminated = true;
+        Some(Err(error))
+    }
+}

crates/io/src/lib.rs

@@ -9,7 +9,8 @@ use uuid::Uuid;
 
 // Re-exports.
 pub use quent_io_types::{
-    Exporter, ExporterProvider, ExporterResult, ImporterError, ImporterProvider, ImporterResult,
+    Exporter, ExporterProvider, ExporterResult, Importer, ImporterError, ImporterProvider,
+    ImporterResult,
 };
 
 // Feature-gated re-exports for convenience.

crates/io/types/src/lib.rs

@@ -82,17 +82,34 @@ impl From<std::io::Error> for ExporterError {
 /// Result of exporters.
 pub type ExporterResult<T> = std::result::Result<T, ExporterError>;
 
-#[derive(Error, Debug)]
+#[derive(Debug, Error)]
 pub enum ImporterError {
-    #[error("i/o error: {0}")]
-    IoError(#[from] std::io::Error),
+    /// Any failure originating in the importer implementation.
+    #[error(transparent)]
+    Other(#[from] Box<dyn std::error::Error + Send + Sync>),
+}
+
+impl ImporterError {
+    /// Wrap an implementation-specific error as [`ImporterError::Other`].
+    pub fn other<E: std::error::Error + Send + Sync + 'static>(error: E) -> Self {
+        Self::Other(Box::new(error))
+    }
 }
 
+impl From<std::io::Error> for ImporterError {
+    fn from(error: std::io::Error) -> Self {
+        Self::other(error)
+    }
+}
+
+/// Maximum supported payload size for length-prefixed importer frames.
+pub const MAX_FRAME_SIZE_BYTES: usize = 64 * 1024 * 1024;
+
 /// Result type for importers.
 pub type ImporterResult<T> = std::result::Result<T, ImporterError>;
 
 /// A source of one entity's events.
-pub trait Importer<T>: Iterator<Item = Event<T>> {}
+pub trait Importer<T>: Iterator<Item = ImporterResult<Event<T>>> {}
 
 /// Provides an importer instance for `T`.
 pub trait ImporterProvider<T> {
@@ -104,8 +121,9 @@ pub trait ImporterProvider<T> {
 /// unchanged.
 ///
 /// # Errors
-/// Returns [`ImporterError::IoError`] if the directory cannot be read or
-/// contains no file with extension `ext`.
+///
+/// Returns an error if the directory cannot be read or contains no file with
+/// extension `ext`.
 pub fn resolve_import_path(
     path: &std::path::Path,
     ext: &str,
@@ -119,10 +137,11 @@ pub fn resolve_import_path(
             return Ok(candidate);
         }
     }
-    Err(ImporterError::IoError(std::io::Error::new(
+    Err(std::io::Error::new(
         std::io::ErrorKind::NotFound,
         format!("no .{ext} file found in directory {}", path.display()),
-    )))
+    )
+    .into())
 }
 
 #[cfg(test)]