From 20dc30ea182de606aa35242192ca4c8f4eae8955 Mon Sep 17 00:00:00 2001
From: rm-dr <96270320+rm-dr@users.noreply.github.com>
Date: Tue, 10 Mar 2026 09:18:26 -0700
Subject: [PATCH] Read files incrementally

---
 .../src/extract/epub/epub_meta.rs             |  58 ++++++---
 .../src/extract/epub/epub_text.rs             |  52 +++++---
 crates/pile-dataset/src/extract/exif.rs       |  42 ++++--
 crates/pile-dataset/src/extract/flac.rs       |  40 +++---
 crates/pile-dataset/src/extract/id3.rs        |  44 +++----
 .../pile-dataset/src/extract/pdf/pdf_meta.rs  | 103 ++++++++-------
 .../pile-dataset/src/extract/pdf/pdf_text.rs  | 102 +++++++++------
 crates/pile-dataset/src/extract/toml.rs       |   2 +-
 crates/pile-dataset/src/item.rs               | 123 +++++++++++++-----
 9 files changed, 352 insertions(+), 214 deletions(-)

diff --git a/crates/pile-dataset/src/extract/epub/epub_meta.rs b/crates/pile-dataset/src/extract/epub/epub_meta.rs
index 4681a67..d6edfa6 100644
--- a/crates/pile-dataset/src/extract/epub/epub_meta.rs
+++ b/crates/pile-dataset/src/extract/epub/epub_meta.rs
@@ -1,9 +1,9 @@
 use epub::doc::EpubDoc;
 use pile_config::Label;
-use std::{collections::HashMap, io::Cursor, sync::OnceLock};
-use tracing::debug;
+use std::{collections::HashMap, sync::OnceLock};
+use tracing::trace;
 
-use crate::{Item, PileValue, extract::Extractor};
+use crate::{Item, PileValue, SyncReadBridge, extract::Extractor};
 
 pub struct EpubMetaExtractor<'a> {
 	item: &'a Item,
@@ -23,34 +23,50 @@ impl<'a> EpubMetaExtractor<'a> {
 			return Ok(x);
 		}
 
-		let bytes = self.item.read().await?.read_to_end().await?;
+		let key = self.item.key();
+		let ext = key.as_str().rsplit('.').next();
+		if !matches!(ext, Some("epub")) {
+			return Ok(self.output.get_or_init(HashMap::new));
+		}
 
-		let cursor = Cursor::new(bytes);
-		let doc = match EpubDoc::from_reader(cursor) {
+		let reader = SyncReadBridge::new_current(self.item.read().await?);
+		let raw_meta = tokio::task::spawn_blocking(move || {
+			let doc = EpubDoc::from_reader(reader)
+				.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
+
+			let fields: &[&'static str] = &[
+				"title",
+				"creator",
+				"description",
+				"language",
+				"publisher",
+				"date",
+				"subject",
+				"identifier",
+			];
+
+			let meta: Vec<(&'static str, Option<String>)> =
+				fields.iter().map(|&key| (key, doc.mdata(key))).collect();
+
+			Ok::<_, std::io::Error>(meta)
+		})
+		.await
+		.map_err(std::io::Error::other)?;
+
+		let raw_meta = match raw_meta {
 			Ok(x) => x,
 			Err(error) => {
-				debug!(message = "Could not process epub", ?error, key = ?self.item.key());
+				trace!(message = "Could not process epub", ?error, key = ?self.item.key());
 				return Ok(self.output.get_or_init(HashMap::new));
 			}
 		};
 
 		let mut output: HashMap<Label, PileValue<'a>> = HashMap::new();
 
-		let fields = &[
-			"title",
-			"creator",
-			"description",
-			"language",
-			"publisher",
-			"date",
-			"subject",
-			"identifier",
-		];
-
 		#[expect(clippy::unwrap_used)]
-		for key in fields {
-			let label = Label::new(*key).unwrap();
-			let value = match doc.mdata(key) {
+		for (key, val) in raw_meta {
+			let label = Label::new(key).unwrap();
+			let value = match val {
 				Some(s) => PileValue::String(s.into()),
 				None => PileValue::Null,
 			};
diff --git a/crates/pile-dataset/src/extract/epub/epub_text.rs b/crates/pile-dataset/src/extract/epub/epub_text.rs
index 8d91c8b..24dbf1a 100644
--- a/crates/pile-dataset/src/extract/epub/epub_text.rs
+++ b/crates/pile-dataset/src/extract/epub/epub_text.rs
@@ -1,9 +1,9 @@
 use epub::doc::EpubDoc;
 use pile_config::Label;
-use std::{collections::HashMap, io::Cursor, sync::OnceLock};
+use std::{collections::HashMap, sync::OnceLock};
 use tracing::debug;
 
-use crate::{Item, PileValue, extract::Extractor};
+use crate::{Item, PileValue, SyncReadBridge, extract::Extractor};
 
 pub struct EpubTextExtractor<'a> {
 	item: &'a Item,
@@ -23,10 +23,34 @@ impl<'a> EpubTextExtractor<'a> {
 			return Ok(x);
 		}
 
-		let bytes = self.item.read().await?.read_to_end().await?;
+		let key = self.item.key();
+		let ext = key.as_str().rsplit('.').next();
+		if !matches!(ext, Some("epub")) {
+			return Ok(self.output.get_or_init(HashMap::new));
+		}
 
-		let cursor = Cursor::new(bytes);
-		let mut doc = match EpubDoc::from_reader(cursor) {
+		let reader = SyncReadBridge::new_current(self.item.read().await?);
+		let raw_text = tokio::task::spawn_blocking(move || {
+			let mut doc = EpubDoc::from_reader(reader)
+				.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
+
+			let mut text_parts: Vec<String> = Vec::new();
+
+			loop {
+				if let Ok(content) = doc.get_current_str() {
+					text_parts.push(strip_html(&content));
+				}
+				if doc.go_next().is_err() {
+					break;
+				}
+			}
+
+			Ok::<_, std::io::Error>(text_parts.join(" "))
+		})
+		.await
+		.map_err(std::io::Error::other)?;
+
+		let raw_text = match raw_text {
 			Ok(x) => x,
 			Err(error) => {
 				debug!(message = "Could not process epub", ?error, key = ?self.item.key());
@@ -34,21 +58,11 @@ impl<'a> EpubTextExtractor<'a> {
 			}
 		};
 
-		let mut text_parts: Vec<String> = Vec::new();
-
-		loop {
-			if let Ok(content) = doc.get_current_str() {
-				text_parts.push(strip_html(&content));
-			}
-			if doc.go_next().is_err() {
-				break;
-			}
-		}
-
-		let text = text_parts.join(" ");
-
 		#[expect(clippy::unwrap_used)]
-		let output = HashMap::from([(Label::new("text").unwrap(), PileValue::String(text.into()))]);
+		let output = HashMap::from([(
+			Label::new("text").unwrap(),
+			PileValue::String(raw_text.into()),
+		)]);
 
 		let _ = self.output.set(output);
 		#[expect(clippy::unwrap_used)]
diff --git a/crates/pile-dataset/src/extract/exif.rs b/crates/pile-dataset/src/extract/exif.rs
index d879b78..d2166f2 100644
--- a/crates/pile-dataset/src/extract/exif.rs
+++ b/crates/pile-dataset/src/extract/exif.rs
@@ -1,8 +1,8 @@
 use pile_config::Label;
-use std::{collections::HashMap, io::Cursor, sync::OnceLock};
+use std::{collections::HashMap, io::BufReader, sync::OnceLock};
 use tracing::debug;
 
-use crate::{Item, PileValue, extract::Extractor};
+use crate::{Item, PileValue, SyncReadBridge, extract::Extractor};
 
 pub struct ExifExtractor<'a> {
 	item: &'a Item,
@@ -22,10 +22,29 @@ impl<'a> ExifExtractor<'a> {
 			return Ok(x);
 		}
 
-		let bytes = self.item.read().await?.read_to_end().await?;
-		let mut cursor = Cursor::new(bytes);
+		let reader = SyncReadBridge::new_current(self.item.read().await?);
+		let raw_fields = tokio::task::spawn_blocking(move || {
+			let mut br = BufReader::new(reader);
+			let exif = exif::Reader::new()
+				.read_from_container(&mut br)
+				.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
 
-		let exif = match exif::Reader::new().read_from_container(&mut cursor) {
+			let fields: Vec<(String, String)> = exif
+				.fields()
+				.map(|f| {
+					(
+						f.tag.to_string(),
+						f.display_value().with_unit(&exif).to_string(),
+					)
+				})
+				.collect();
+
+			Ok::<_, std::io::Error>(fields)
+		})
+		.await
+		.map_err(std::io::Error::other)?;
+
+		let raw_fields = match raw_fields {
 			Ok(x) => x,
 			Err(error) => {
 				debug!(message = "Could not process exif", ?error, key = ?self.item.key());
@@ -35,23 +54,22 @@ impl<'a> ExifExtractor<'a> {
 
 		let mut output: HashMap<Label, PileValue<'a>> = HashMap::new();
 
-		for field in exif.fields() {
-			let Some(label) = tag_to_label(&field.tag) else {
+		for (tag_name, value) in raw_fields {
+			let Some(label) = tag_to_label(&tag_name) else {
 				continue;
 			};
 			// First occurrence wins (PRIMARY IFD comes before THUMBNAIL)
-			output.entry(label).or_insert_with(|| {
-				PileValue::String(field.display_value().with_unit(&exif).to_string().into())
-			});
+			output
+				.entry(label)
+				.or_insert_with(|| PileValue::String(value.into()));
 		}
 
 		return Ok(self.output.get_or_init(|| output));
 	}
 }
 
-fn tag_to_label(tag: &exif::Tag) -> Option<Label> {
+fn tag_to_label(tag: &str) -> Option<Label> {
 	let sanitized: String = tag
-		.to_string()
 		.chars()
 		.map(|c| if c == ' ' { '_' } else { c })
 		.filter(|c| Label::VALID_CHARS.contains(*c))
diff --git a/crates/pile-dataset/src/extract/flac.rs b/crates/pile-dataset/src/extract/flac.rs
index 47b0ef1..0a42f01 100644
--- a/crates/pile-dataset/src/extract/flac.rs
+++ b/crates/pile-dataset/src/extract/flac.rs
@@ -2,7 +2,7 @@ use pile_config::Label;
 use pile_flac::{FlacBlock, FlacReader};
 use std::{collections::HashMap, io::BufReader, sync::OnceLock};
 
-use crate::{Item, PileValue, extract::Extractor};
+use crate::{Item, PileValue, SyncReadBridge, extract::Extractor};
 
 pub struct FlacExtractor<'a> {
 	item: &'a Item,
@@ -33,24 +33,32 @@ impl<'a> FlacExtractor<'a> {
 			return Ok(self.output.get().unwrap());
 		}
 
-		let bytes = self.item.read().await?.read_to_end().await?;
-		let reader = FlacReader::new(BufReader::new(std::io::Cursor::new(bytes)));
-
-		let mut output: HashMap<Label, Vec<_>> = HashMap::new();
-		for block in reader {
-			if let FlacBlock::VorbisComment(comment) =
-				block.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?
-			{
-				for (k, v) in comment.comment.comments {
-					match Label::new(k.to_string().to_lowercase()) {
-						Some(k) => output.entry(k).or_default().push(PileValue::String(v)),
-						None => continue,
+		let reader = SyncReadBridge::new_current(self.item.read().await?);
+		let raw_tags = tokio::task::spawn_blocking(move || {
+			let reader = FlacReader::new(BufReader::new(reader));
+			let mut tags: Vec<(String, String)> = Vec::new();
+			for block in reader {
+				if let FlacBlock::VorbisComment(comment) =
+					block.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?
+				{
+					for (k, v) in comment.comment.comments {
+						tags.push((k.to_string().to_lowercase(), v.into()));
 					}
+					break;
 				}
+			}
+			Ok::<_, std::io::Error>(tags)
+		})
+		.await
+		.map_err(std::io::Error::other)??;
 
-				// We should only have one comment block,
-				// stop reading when we find it
-				break;
+		let mut output: HashMap<Label, Vec<PileValue<'a>>> = HashMap::new();
+		for (k, v) in raw_tags {
+			if let Some(label) = Label::new(k) {
+				output
+					.entry(label)
+					.or_default()
+					.push(PileValue::String(v.into()));
 			}
 		}
 
diff --git a/crates/pile-dataset/src/extract/id3.rs b/crates/pile-dataset/src/extract/id3.rs
index 22a41e5..94455cc 100644
--- a/crates/pile-dataset/src/extract/id3.rs
+++ b/crates/pile-dataset/src/extract/id3.rs
@@ -2,7 +2,7 @@ use id3::Tag;
 use pile_config::Label;
 use std::{borrow::Cow, collections::HashMap, io::BufReader, sync::OnceLock};
 
-use crate::{Item, PileValue, extract::Extractor};
+use crate::{Item, PileValue, SyncReadBridge, extract::Extractor};
 
 pub struct Id3Extractor<'a> {
 	item: &'a Item,
@@ -22,34 +22,32 @@ impl<'a> Id3Extractor<'a> {
 			return Ok(x);
 		}
 
-		let key = match self.item {
-			Item::File { path, .. } => path.to_str().unwrap_or_default().to_owned(),
-			Item::S3 { key, .. } => key.to_string(),
-		};
-
-		let ext = key.rsplit('.').next();
+		let key = self.item.key();
+		let ext = key.as_str().rsplit('.').next();
 		if !matches!(ext, Some("mp3") | Some("aiff") | Some("aif") | Some("wav")) {
-			let _ = self.output.set(HashMap::new());
-			#[expect(clippy::unwrap_used)]
-			return Ok(self.output.get().unwrap());
+			return Ok(self.output.get_or_init(HashMap::new));
 		}
 
-		let bytes = self.item.read().await?.read_to_end().await?;
-		let tag = match Tag::read_from2(BufReader::new(std::io::Cursor::new(bytes))) {
-			Ok(tag) => tag,
-			Err(id3::Error {
+		let reader = SyncReadBridge::new_current(self.item.read().await?);
+		let tag = match tokio::task::spawn_blocking(move || Tag::read_from2(BufReader::new(reader)))
+			.await
+		{
+			Ok(Ok(tag)) => tag,
+
+			Ok(Err(id3::Error {
 				kind: id3::ErrorKind::NoTag,
 				..
-			}) => {
-				let _ = self.output.set(HashMap::new());
-				#[expect(clippy::unwrap_used)]
-				return Ok(self.output.get().unwrap());
+			})) => {
+				return Ok(self.output.get_or_init(HashMap::new));
 			}
-			Err(id3::Error {
+
+			Ok(Err(id3::Error {
 				kind: id3::ErrorKind::Io(e),
 				..
-			}) => return Err(e),
-			Err(e) => return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e)),
+			})) => return Err(e),
+
+			Ok(Err(e)) => return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e)),
+			Err(e) => return Err(e.into()),
 		};
 
 		let mut output: HashMap<Label, Vec<PileValue<'a>>> = HashMap::new();
@@ -70,9 +68,7 @@ impl<'a> Id3Extractor<'a> {
 			.map(|(k, v)| (k, PileValue::Array(v)))
 			.collect();
 
-		let _ = self.output.set(output);
-		#[expect(clippy::unwrap_used)]
-		return Ok(self.output.get().unwrap());
+		return Ok(self.output.get_or_init(|| output));
 	}
 }
 
diff --git a/crates/pile-dataset/src/extract/pdf/pdf_meta.rs b/crates/pile-dataset/src/extract/pdf/pdf_meta.rs
index 332ebff..d2e927a 100644
--- a/crates/pile-dataset/src/extract/pdf/pdf_meta.rs
+++ b/crates/pile-dataset/src/extract/pdf/pdf_meta.rs
@@ -1,10 +1,10 @@
 use pdf::file::FileOptions;
-use pdf::primitive::{Date, PdfString, TimeRel};
+use pdf::primitive::{Date, TimeRel};
 use pile_config::Label;
-use std::{collections::HashMap, sync::OnceLock};
-use tracing::debug;
+use std::{collections::HashMap, io::BufReader, sync::OnceLock};
+use tracing::trace;
 
-use crate::{Item, PileValue, extract::Extractor};
+use crate::{Item, PileValue, SyncReadBridge, extract::Extractor};
 
 pub struct PdfMetaExtractor<'a> {
 	item: &'a Item,
@@ -24,56 +24,69 @@ impl<'a> PdfMetaExtractor<'a> {
 			return Ok(x);
 		}
 
-		let bytes = self.item.read().await?.read_to_end().await?;
+		let reader = SyncReadBridge::new_current(self.item.read().await?);
+		let raw_meta = tokio::task::spawn_blocking(move || {
+			let mut bytes = Vec::new();
+			std::io::Read::read_to_end(&mut BufReader::new(reader), &mut bytes)?;
 
-		let file = match FileOptions::cached().load(bytes) {
+			let file = match FileOptions::cached().load(bytes) {
+				Ok(x) => x,
+				Err(pdf::PdfError::Io { source }) => return Err(source),
+				Err(error) => {
+					return Err(std::io::Error::new(
+						std::io::ErrorKind::InvalidData,
+						error.to_string(),
+					));
+				}
+			};
+
+			let mut meta: Vec<(&'static str, Option<String>)> = Vec::new();
+
+			if let Some(info) = &file.trailer.info_dict {
+				use pdf::primitive::PdfString;
+				let fields: &[(&'static str, Option<&PdfString>)] = &[
+					("title", info.title.as_ref()),
+					("author", info.author.as_ref()),
+					("subject", info.subject.as_ref()),
+					("keywords", info.keywords.as_ref()),
+					("creator", info.creator.as_ref()),
+					("producer", info.producer.as_ref()),
+				];
+
+				for (key, val) in fields {
+					meta.push((key, val.map(|s| s.to_string_lossy())));
+				}
+
+				meta.push((
+					"creation_date",
+					info.creation_date.as_ref().map(format_date),
+				));
+				meta.push(("mod_date", info.mod_date.as_ref().map(format_date)));
+			}
+
+			Ok::<_, std::io::Error>(meta)
+		})
+		.await
+		.map_err(std::io::Error::other)?;
+
+		let raw_meta = match raw_meta {
 			Ok(x) => x,
-			Err(pdf::PdfError::Io { source }) => return Err(source),
 			Err(error) => {
-				debug!(message = "Could not process pdf", ?error, key = ?self.item.key());
+				trace!(message = "Could not process pdf", ?error, key = ?self.item.key());
 				return Ok(self.output.get_or_init(HashMap::new));
 			}
 		};
 
 		let mut output: HashMap<Label, PileValue<'a>> = HashMap::new();
 
-		if let Some(info) = &file.trailer.info_dict {
-			let fields: &[(&str, Option<&PdfString>)] = &[
-				("title", info.title.as_ref()),
-				("author", info.author.as_ref()),
-				("subject", info.subject.as_ref()),
-				("keywords", info.keywords.as_ref()),
-				("creator", info.creator.as_ref()),
-				("producer", info.producer.as_ref()),
-			];
-
-			#[expect(clippy::unwrap_used)]
-			for (key, val) in fields {
-				let label = Label::new(*key).unwrap();
-				let value = match val {
-					Some(s) => PileValue::String(s.to_string_lossy().into()),
-					None => PileValue::Null,
-				};
-				output.insert(label, value);
-			}
-
-			#[expect(clippy::unwrap_used)]
-			{
-				output.insert(
-					Label::new("creation_date").unwrap(),
-					info.creation_date
-						.as_ref()
-						.map(|d| PileValue::String(format_date(d).into()))
-						.unwrap_or(PileValue::Null),
-				);
-				output.insert(
-					Label::new("mod_date").unwrap(),
-					info.mod_date
-						.as_ref()
-						.map(|d| PileValue::String(format_date(d).into()))
-						.unwrap_or(PileValue::Null),
-				);
-			}
+		#[expect(clippy::unwrap_used)]
+		for (key, val) in raw_meta {
+			let label = Label::new(key).unwrap();
+			let value = match val {
+				Some(s) => PileValue::String(s.into()),
+				None => PileValue::Null,
+			};
+			output.insert(label, value);
 		}
 
 		return Ok(self.output.get_or_init(|| output));
diff --git a/crates/pile-dataset/src/extract/pdf/pdf_text.rs b/crates/pile-dataset/src/extract/pdf/pdf_text.rs
index 1c707d6..7c2f37f 100644
--- a/crates/pile-dataset/src/extract/pdf/pdf_text.rs
+++ b/crates/pile-dataset/src/extract/pdf/pdf_text.rs
@@ -1,10 +1,10 @@
 use pdf::content::{Op, TextDrawAdjusted};
 use pdf::file::FileOptions;
 use pile_config::Label;
-use std::{collections::HashMap, sync::OnceLock};
-use tracing::debug;
+use std::{collections::HashMap, io::BufReader, sync::OnceLock};
+use tracing::trace;
 
-use crate::{Item, PileValue, extract::Extractor};
+use crate::{Item, PileValue, SyncReadBridge, extract::Extractor};
 
 pub struct PdfTextExtractor<'a> {
 	item: &'a Item,
@@ -24,54 +24,72 @@ impl<'a> PdfTextExtractor<'a> {
 			return Ok(x);
 		}
 
-		let bytes = self.item.read().await?.read_to_end().await?;
+		let reader = SyncReadBridge::new_current(self.item.read().await?);
+		let raw_text = tokio::task::spawn_blocking(move || {
+			let mut bytes = Vec::new();
+			std::io::Read::read_to_end(&mut BufReader::new(reader), &mut bytes)?;
 
-		let file = match FileOptions::cached().load(bytes) {
+			let file = match FileOptions::cached().load(bytes) {
+				Ok(x) => x,
+				Err(pdf::PdfError::Io { source }) => return Err(source),
+				Err(error) => {
+					return Err(std::io::Error::new(
+						std::io::ErrorKind::InvalidData,
+						error.to_string(),
+					));
+				}
+			};
+
+			let mut text_parts: Vec<String> = Vec::new();
+
+			for page in file.pages() {
+				let page = page.map_err(|e| {
+					std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())
+				})?;
+
+				if let Some(content) = &page.contents {
+					let ops = content.operations(&file.resolver()).map_err(|e| {
+						std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())
+					})?;
+
+					for op in ops {
+						match op {
+							Op::TextDraw { text } => {
+								text_parts.push(text.to_string_lossy());
+							}
+							Op::TextDrawAdjusted { array } => {
+								for item in array {
+									if let TextDrawAdjusted::Text(text) = item {
+										text_parts.push(text.to_string_lossy());
+									}
+								}
+							}
+							_ => {}
+						}
+					}
+				}
+			}
+
+			Ok::<_, std::io::Error>(text_parts.join(" "))
+		})
+		.await
+		.map_err(std::io::Error::other)?;
+
+		let raw_text = match raw_text {
 			Ok(x) => x,
-			Err(pdf::PdfError::Io { source }) => return Err(source),
 			Err(error) => {
-				debug!(message = "Could not process pdf", ?error, key = ?self.item.key());
+				trace!(message = "Could not process pdf", ?error, key = ?self.item.key());
 				return Ok(self.output.get_or_init(HashMap::new));
 			}
 		};
 
-		let mut text_parts: Vec<String> = Vec::new();
-
-		for page in file.pages() {
-			let page = page
-				.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string()))?;
-
-			if let Some(content) = &page.contents {
-				let ops = content.operations(&file.resolver()).map_err(|e| {
-					std::io::Error::new(std::io::ErrorKind::InvalidData, e.to_string())
-				})?;
-
-				for op in ops {
-					match op {
-						Op::TextDraw { text } => {
-							text_parts.push(text.to_string_lossy());
-						}
-						Op::TextDrawAdjusted { array } => {
-							for item in array {
-								if let TextDrawAdjusted::Text(text) = item {
-									text_parts.push(text.to_string_lossy());
-								}
-							}
-						}
-						_ => {}
-					}
-				}
-			}
-		}
-
-		let text = text_parts.join(" ");
-
 		#[expect(clippy::unwrap_used)]
-		let output = HashMap::from([(Label::new("text").unwrap(), PileValue::String(text.into()))]);
+		let output = HashMap::from([(
+			Label::new("text").unwrap(),
+			PileValue::String(raw_text.into()),
+		)]);
 
-		let _ = self.output.set(output);
-		#[expect(clippy::unwrap_used)]
-		return Ok(self.output.get().unwrap());
+		return Ok(self.output.get_or_init(|| output));
 	}
 }
 
diff --git a/crates/pile-dataset/src/extract/toml.rs b/crates/pile-dataset/src/extract/toml.rs
index 40d608e..4713084 100644
--- a/crates/pile-dataset/src/extract/toml.rs
+++ b/crates/pile-dataset/src/extract/toml.rs
@@ -1,7 +1,7 @@
 use pile_config::Label;
 use std::{collections::HashMap, sync::OnceLock};
 
-use crate::{Item, PileValue, extract::Extractor};
+use crate::{Item, PileValue, Reader, extract::Extractor};
 
 fn toml_to_pile(value: toml::Value) -> PileValue<'static> {
 	match value {
diff --git a/crates/pile-dataset/src/item.rs b/crates/pile-dataset/src/item.rs
index 3fcc3c5..acaa664 100644
--- a/crates/pile-dataset/src/item.rs
+++ b/crates/pile-dataset/src/item.rs
@@ -1,5 +1,11 @@
 use smartstring::{LazyCompact, SmartString};
-use std::{fs::File, io::Seek, path::PathBuf, sync::Arc};
+use std::{
+	fs::File,
+	io::{Read, Seek, SeekFrom},
+	path::PathBuf,
+	sync::Arc,
+};
+use tokio::runtime::Handle;
 
 use crate::source::{DirDataSource, S3DataSource};
 
@@ -90,47 +96,96 @@ impl Item {
 	}
 }
 
+//
+// MARK: reader
+//
+
+pub trait Reader: Send {
+	/// Read a chunk of bytes.
+	fn read(
+		&mut self,
+		buf: &mut [u8],
+	) -> impl Future<Output = Result<usize, std::io::Error>> + Send;
+
+	fn seek(&mut self, pos: SeekFrom) -> impl Future<Output = Result<u64, std::io::Error>> + Send;
+
+	/// Read all remaining bytes into a `Vec`.
+	fn read_to_end(&mut self) -> impl Future<Output = Result<Vec<u8>, std::io::Error>> + Send {
+		async {
+			let mut buf = Vec::new();
+			let mut chunk = vec![0u8; 65536];
+			loop {
+				let n = self.read(&mut chunk).await?;
+				if n == 0 {
+					break;
+				}
+				buf.extend_from_slice(&chunk[..n]);
+			}
+			Ok(buf)
+		}
+	}
+}
+
+//
+// MARK: sync bridge
+//
+
+/// Turn an async [Reader] into a sync [Read] + [Seek].
+///
+/// Never use this outside of [tokio::task::spawn_blocking],
+/// the async runtime will deadlock if this struct blocks
+/// the runtime.
+pub struct SyncReadBridge<R: Reader> {
+	inner: R,
+	handle: Handle,
+}
+
+impl<R: Reader> SyncReadBridge<R> {
+	/// Creates a new adapter using a handle to the current runtime.
+	/// Panics if called outside of tokio
+	pub fn new_current(inner: R) -> Self {
+		Self::new(inner, Handle::current())
+	}
+
+	/// Creates a new adapter using a handle to an existing runtime.
+	pub fn new(inner: R, handle: Handle) -> Self {
+		Self { inner, handle }
+	}
+}
+
+impl<R: Reader> Read for SyncReadBridge<R> {
+	fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
+		self.handle.block_on(self.inner.read(buf))
+	}
+}
+
+impl<R: Reader> Seek for SyncReadBridge<R> {
+	fn seek(&mut self, pos: SeekFrom) -> Result<u64, std::io::Error> {
+		self.handle.block_on(self.inner.seek(pos))
+	}
+}
+
+//
+// MARK: itemreader
+//
+
 pub enum ItemReader {
 	File(File),
 	S3(S3Reader),
 }
 
-impl ItemReader {
-	/// Read a chunk of bytes.
-	pub async fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+impl Reader for ItemReader {
+	async fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
 		match self {
 			Self::File(x) => std::io::Read::read(x, buf),
 			Self::S3(x) => x.read(buf).await,
 		}
 	}
 
-	/// Read all remaining bytes into a `Vec`.
-	pub async fn read_to_end(mut self) -> std::io::Result<Vec<u8>> {
-		match self {
-			Self::File(mut f) => {
-				let mut buf = Vec::new();
-				std::io::Read::read_to_end(&mut f, &mut buf)?;
-				Ok(buf)
-			}
-			Self::S3(ref mut r) => {
-				let mut buf = Vec::new();
-				let mut chunk = vec![0u8; 65536];
-				loop {
-					let n = r.read(&mut chunk).await?;
-					if n == 0 {
-						break;
-					}
-					buf.extend_from_slice(&chunk[..n]);
-				}
-				Ok(buf)
-			}
-		}
-	}
-
-	pub fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
+	async fn seek(&mut self, pos: std::io::SeekFrom) -> Result<u64, std::io::Error> {
 		match self {
 			Self::File(x) => x.seek(pos),
-			Self::S3(x) => x.seek(pos),
+			Self::S3(x) => x.seek(pos).await,
 		}
 	}
 }
@@ -147,8 +202,8 @@ pub struct S3Reader {
 	size: u64,
 }
 
-impl S3Reader {
-	async fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+impl Reader for S3Reader {
+	async fn read(&mut self, buf: &mut [u8]) -> Result<usize, std::io::Error> {
 		let len_left = self.size.saturating_sub(self.cursor);
 		if len_left == 0 || buf.is_empty() {
 			return Ok(0);
@@ -181,11 +236,11 @@ impl S3Reader {
 		Ok(n)
 	}
 
-	fn seek(&mut self, pos: std::io::SeekFrom) -> std::io::Result<u64> {
+	async fn seek(&mut self, pos: SeekFrom) -> Result<u64, std::io::Error> {
 		match pos {
-			std::io::SeekFrom::Start(x) => self.cursor = x.min(self.size),
+			SeekFrom::Start(x) => self.cursor = x.min(self.size),
 
-			std::io::SeekFrom::Current(x) => {
+			SeekFrom::Current(x) => {
 				if x < 0 {
 					let abs = x.unsigned_abs();
 					if abs > self.cursor {