diff --git a/crates/io/msgpack/src/lib.rs b/crates/io/msgpack/src/lib.rs
index 412fbb111..1fd659e67 100644
--- a/crates/io/msgpack/src/lib.rs
+++ b/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))
+    }
+}
diff --git a/crates/io/ndjson/src/lib.rs b/crates/io/ndjson/src/lib.rs
index d4516d292..da9416c85 100644
--- a/crates/io/ndjson/src/lib.rs
+++ b/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()))
             }
         }
     }
diff --git a/crates/io/postcard/src/lib.rs b/crates/io/postcard/src/lib.rs
index 0dc70aa9d..fbd7c21d5 100644
--- a/crates/io/postcard/src/lib.rs
+++ b/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))
+    }
+}
diff --git a/crates/io/src/lib.rs b/crates/io/src/lib.rs
index d73f6aaa8..9732722cd 100644
--- a/crates/io/src/lib.rs
+++ b/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.
diff --git a/crates/io/types/src/lib.rs b/crates/io/types/src/lib.rs
index 9399bcbe4..fba987c5e 100644
--- a/crates/io/types/src/lib.rs
+++ b/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)]
diff --git a/crates/model-macros/src/model_macro.rs b/crates/model-macros/src/model_macro.rs
index d72897fac..5b25a47fa 100644
--- a/crates/model-macros/src/model_macro.rs
+++ b/crates/model-macros/src/model_macro.rs
@@ -353,7 +353,13 @@ pub fn expand(input: TokenStream) -> syn::Result<TokenStream> {
                 pub fn import_events(
                     dir: &std::path::Path,
                 ) -> quent_model::io::ImporterResult<
-                    Box<dyn Iterator<Item = quent_model::Event<#event_type>>>,
+                    Box<
+                        dyn Iterator<
+                            Item = quent_model::io::ImporterResult<
+                                quent_model::Event<#event_type>
+                            >,
+                        >,
+                    >,
                 > {
                     // Detect the on-disk serialization format from the streams present;
                     // an empty/unrecognized context yields no events.
@@ -361,7 +367,13 @@ pub fn expand(input: TokenStream) -> syn::Result<TokenStream> {
                         return Ok(Box::new(std::iter::empty()));
                     };
                     let mut streams: Vec<
-                        Box<dyn Iterator<Item = quent_model::Event<#event_type>>>,
+                        Box<
+                            dyn Iterator<
+                                Item = quent_model::io::ImporterResult<
+                                    quent_model::Event<#event_type>
+                                >,
+                            >,
+                        >,
                     > = Vec::new();
                     #(
                         {
@@ -376,12 +388,14 @@ pub fn expand(input: TokenStream) -> syn::Result<TokenStream> {
                                         },
                                     ),
                                 )?;
-                                streams.push(Box::new(importer.map(|e| {
-                                    quent_model::Event::new(
-                                        e.id,
-                                        e.timestamp,
-                                        #event_type::from(e.data),
-                                    )
+                                streams.push(Box::new(importer.map(|event| {
+                                    event.map(|event| {
+                                        quent_model::Event::new(
+                                            event.id,
+                                            event.timestamp,
+                                            #event_type::from(event.data),
+                                        )
+                                    })
                                 })));
                             }
                         }
diff --git a/domains/query_engine/server/src/analyzer_cache.rs b/domains/query_engine/server/src/analyzer_cache.rs
index b805f4b09..cb355bd08 100644
--- a/domains/query_engine/server/src/analyzer_cache.rs
+++ b/domains/query_engine/server/src/analyzer_cache.rs
@@ -114,6 +114,7 @@ pub fn index_query_engines(output_dir: &Path) -> ServerResult<EngineIndex> {
             )?;
             let mut seen = HashSet::new();
             for event in importer {
+                let event = event?;
                 if seen.insert(event.id) {
                     index.attribute_context(event.id, context_id);
                 }
@@ -131,6 +132,7 @@ pub fn index_query_engines(output_dir: &Path) -> ServerResult<EngineIndex> {
             )?;
             let mut seen = HashSet::new();
             for event in importer {
+                let event = event?;
                 if let WorkerEvent::Init(init) = &event.data
                     && seen.insert(event.id)
                 {
diff --git a/examples/simulator/analyzer/src/lib.rs b/examples/simulator/analyzer/src/lib.rs
index 123dfdaa9..10a37fa0c 100644
--- a/examples/simulator/analyzer/src/lib.rs
+++ b/examples/simulator/analyzer/src/lib.rs
@@ -154,7 +154,9 @@ impl QuentViewer for Viewer {
     fn import_events(
         dir: &std::path::Path,
     ) -> quent_model::io::ImporterResult<ViewerEventStream<Self::Analyzer>> {
-        Simulator::import_events(dir)
+        let events =
+            Simulator::import_events(dir)?.collect::<quent_model::io::ImporterResult<Vec<_>>>()?;
+        Ok(Box::new(events.into_iter()))
     }
 }
 
diff --git a/examples/simulator/server/src/main.rs b/examples/simulator/server/src/main.rs
index 050c8a3a5..b367d5eff 100644
--- a/examples/simulator/server/src/main.rs
+++ b/examples/simulator/server/src/main.rs
@@ -121,7 +121,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // make up an engine instance.
     let importer = move |context_id| {
         let dir = importer_output_dir.join(format!("{context_id}"));
-        Ok(Simulator::import_events(&dir)?)
+        let events =
+            Simulator::import_events(&dir)?.collect::<quent_io::ImporterResult<Vec<_>>>()?;
+        Ok::<Box<dyn Iterator<Item = _>>, quent_query_engine_server::error::ServerError>(Box::new(
+            events.into_iter(),
+        ))
     };
 
     let analyzer = async {