Implement content-addressed resource storage for raster images (#4148)

* Implement content-addressed resource storage

* Implement OPFS resource storage

* Remove ResourceStorage::read

* Review

node-graph/graph-craft/Cargo.toml

@@ -50,12 +50,26 @@ wasm-bindgen = { workspace = true, optional = true }
 
 # Workspace dependencies
 [target.'cfg(target_family = "wasm")'.dependencies]
-web-sys = { workspace = true, features = ["Navigator", "Gpu"] }
+web-sys = { workspace = true, features = [
+	"Navigator",
+	"DomException",
+	"Window",
+	"StorageManager",
+	"FileSystemDirectoryHandle",
+	"FileSystemFileHandle",
+	"FileSystemGetFileOptions",
+	"FileSystemGetDirectoryOptions",
+	"FileSystemWritableFileStream",
+	"WritableStream",
+	"Blob",
+] }
 js-sys = { workspace = true }
 wasm-bindgen = { workspace = true }
+wasm-bindgen-futures = { workspace = true }
 
 [target.'cfg(not(target_family = "wasm"))'.dependencies]
 winit = { workspace = true }
+mmap-io = { workspace = true }
 
 [dev-dependencies]
 # Workspace dependencies

node-graph/graph-craft/src/application_io.rs

@@ -1,14 +1,84 @@
 use dyn_any::StaticType;
+#[cfg(feature = "wgpu")]
+use wgpu_executor::WgpuExecutor;
 
-#[cfg(not(target_family = "wasm"))]
-mod native;
-#[cfg(target_family = "wasm")]
-mod wasm;
+pub mod resource;
 
+pub use graphene_application_io::{ApplicationIo, LoadResource, Resource, ResourceFuture, ResourceHash, ResourceStorage};
+pub use resource::HashMapResourceStorage;
 #[cfg(not(target_family = "wasm"))]
-pub type PlatformApplicationIo = native::NativeApplicationIo;
+pub use resource::mmap::MmapResourceStorage;
 #[cfg(target_family = "wasm")]
-pub type PlatformApplicationIo = wasm::WasmApplicationIo;
+pub use resource::opfs::OpfsResourceStorage;
+
+#[derive(Default)]
+pub struct PlatformApplicationIo {
+	#[cfg(feature = "wgpu")]
+	pub(crate) gpu_executor: Option<WgpuExecutor>,
+	resources: Option<Box<dyn LoadResource>>,
+}
+
+impl PlatformApplicationIo {
+	pub async fn new() -> Self {
+		#[cfg(feature = "wgpu")]
+		let executor = WgpuExecutor::new().await;
+
+		#[cfg(not(feature = "wgpu"))]
+		let wgpu_available = false;
+		#[cfg(feature = "wgpu")]
+		let wgpu_available = executor.is_some();
+		set_wgpu_available(wgpu_available);
+
+		Self {
+			#[cfg(feature = "wgpu")]
+			gpu_executor: executor,
+			resources: None,
+		}
+	}
+
+	#[cfg(feature = "wgpu")]
+	pub fn new_with_context(context: wgpu_executor::WgpuContext) -> Self {
+		let executor = WgpuExecutor::with_context(context);
+
+		let wgpu_available = executor.is_some();
+		set_wgpu_available(wgpu_available);
+
+		Self {
+			gpu_executor: executor,
+			resources: None,
+		}
+	}
+
+	pub fn inject_resource_proxy(&mut self, resources: Box<dyn LoadResource>) {
+		self.resources = Some(resources);
+	}
+}
+
+impl ApplicationIo for PlatformApplicationIo {
+	#[cfg(feature = "wgpu")]
+	type Executor = WgpuExecutor;
+	#[cfg(not(feature = "wgpu"))]
+	type Executor = ();
+
+	#[cfg(feature = "wgpu")]
+	fn gpu_executor(&self) -> Option<&Self::Executor> {
+		self.gpu_executor.as_ref()
+	}
+
+	fn load_resource(&self, hash: ResourceHash) -> graphene_application_io::ResourceFuture {
+		self.resources.as_ref().expect("Resource storage not initialized").load(hash)
+	}
+}
+
+impl std::fmt::Debug for PlatformApplicationIo {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		f.debug_struct("PlatformApplicationIo").finish_non_exhaustive()
+	}
+}
+
+unsafe impl StaticType for PlatformApplicationIo {
+	type Static = PlatformApplicationIo;
+}
 
 pub type PlatformEditorApi = graphene_application_io::EditorApi<PlatformApplicationIo>;
 

node-graph/graph-craft/src/application_io/native.rs

@@ -1,113 +0,0 @@
-use dyn_any::StaticType;
-use graphene_application_io::{ApplicationError, ApplicationIo, ResourceFuture};
-use std::collections::HashMap;
-use std::sync::Arc;
-#[cfg(feature = "tokio")]
-use tokio::io::AsyncReadExt;
-#[cfg(target_family = "wasm")]
-use wasm_bindgen::JsCast;
-#[cfg(feature = "wgpu")]
-use wgpu_executor::WgpuExecutor;
-
-#[derive(Debug, Default)]
-pub struct NativeApplicationIo {
-	#[cfg(feature = "wgpu")]
-	pub(crate) gpu_executor: Option<WgpuExecutor>,
-	pub resources: HashMap<String, Arc<[u8]>>,
-}
-
-impl NativeApplicationIo {
-	pub async fn new() -> Self {
-		#[cfg(feature = "wgpu")]
-		let executor = WgpuExecutor::new().await;
-
-		#[cfg(not(feature = "wgpu"))]
-		let wgpu_available = false;
-		#[cfg(feature = "wgpu")]
-		let wgpu_available = executor.is_some();
-		super::set_wgpu_available(wgpu_available);
-
-		let mut io = Self {
-			#[cfg(feature = "wgpu")]
-			gpu_executor: executor,
-			resources: HashMap::new(),
-		};
-		io.resources.insert("null".to_string(), Arc::from(include_bytes!("../null.png").to_vec()));
-
-		io
-	}
-
-	#[cfg(feature = "wgpu")]
-	pub fn new_with_context(context: wgpu_executor::WgpuContext) -> Self {
-		#[cfg(feature = "wgpu")]
-		let executor = WgpuExecutor::with_context(context);
-
-		#[cfg(not(feature = "wgpu"))]
-		let wgpu_available = false;
-		#[cfg(feature = "wgpu")]
-		let wgpu_available = executor.is_some();
-		super::set_wgpu_available(wgpu_available);
-
-		let mut io = Self {
-			gpu_executor: executor,
-			resources: HashMap::new(),
-		};
-
-		io.resources.insert("null".to_string(), Arc::from(include_bytes!("../null.png").to_vec()));
-
-		io
-	}
-}
-
-impl ApplicationIo for NativeApplicationIo {
-	#[cfg(feature = "wgpu")]
-	type Executor = WgpuExecutor;
-	#[cfg(not(feature = "wgpu"))]
-	type Executor = ();
-
-	#[cfg(feature = "wgpu")]
-	fn gpu_executor(&self) -> Option<&Self::Executor> {
-		self.gpu_executor.as_ref()
-	}
-
-	fn load_resource(&self, url: impl AsRef<str>) -> Result<ResourceFuture, ApplicationError> {
-		let url = url::Url::parse(url.as_ref()).map_err(|_| ApplicationError::InvalidUrl)?;
-		log::trace!("Loading resource: {url:?}");
-		match url.scheme() {
-			#[cfg(feature = "tokio")]
-			"file" => {
-				let path = url.to_file_path().map_err(|_| ApplicationError::NotFound)?;
-				let path = path.to_str().ok_or(ApplicationError::NotFound)?;
-				let path = path.to_owned();
-				Ok(Box::pin(async move {
-					let file = tokio::fs::File::open(path).await.map_err(|_| ApplicationError::NotFound)?;
-					let mut reader = tokio::io::BufReader::new(file);
-					let mut data = Vec::new();
-					reader.read_to_end(&mut data).await.map_err(|_| ApplicationError::NotFound)?;
-					Ok(Arc::from(data))
-				}) as ResourceFuture)
-			}
-			"http" | "https" => {
-				let url = url.to_string();
-				Ok(Box::pin(async move {
-					let client = reqwest::Client::new();
-					let response = client.get(url).send().await.map_err(|_| ApplicationError::NotFound)?;
-					let data = response.bytes().await.map_err(|_| ApplicationError::NotFound)?;
-					Ok(Arc::from(data.to_vec()))
-				}) as ResourceFuture)
-			}
-			"graphite" => {
-				let path = url.path();
-				let path = path.to_owned();
-				log::trace!("Loading local resource: {path}");
-				let data = self.resources.get(&path).ok_or(ApplicationError::NotFound)?.clone();
-				Ok(Box::pin(async move { Ok(data.clone()) }) as ResourceFuture)
-			}
-			_ => Err(ApplicationError::NotFound),
-		}
-	}
-}
-
-unsafe impl StaticType for NativeApplicationIo {
-	type Static = NativeApplicationIo;
-}

node-graph/graph-craft/src/application_io/resource.rs

@@ -0,0 +1,43 @@
+#[cfg(not(target_family = "wasm"))]
+pub mod mmap;
+#[cfg(target_family = "wasm")]
+pub mod opfs;
+
+use graphene_application_io::{LoadResource, Resource, ResourceFuture, ResourceHash, ResourceStorage};
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+#[derive(Debug, Default)]
+pub struct HashMapResourceStorage {
+	resources: Mutex<HashMap<ResourceHash, Resource>>,
+}
+
+impl HashMapResourceStorage {
+	pub fn new() -> Self {
+		Self::default()
+	}
+}
+
+impl LoadResource for HashMapResourceStorage {
+	fn load(&self, hash: ResourceHash) -> ResourceFuture {
+		let result = self.resources.lock().unwrap().get(&hash).cloned();
+		Box::pin(async move { result })
+	}
+}
+
+impl ResourceStorage for HashMapResourceStorage {
+	fn store(&mut self, data: &[u8]) -> ResourceHash {
+		let hash = ResourceHash::from(data);
+		self.resources.get_mut().unwrap().insert(hash, Resource::new(Arc::<[u8]>::from(data)));
+		hash
+	}
+
+	fn contains(&mut self, hash: &ResourceHash) -> bool {
+		self.resources.get_mut().unwrap().contains_key(hash)
+	}
+
+	fn garbage_collect(&mut self, used: &[ResourceHash]) {
+		let used_set: std::collections::HashSet<&ResourceHash> = used.iter().collect();
+		self.resources.get_mut().unwrap().retain(|hash, _| used_set.contains(hash));
+	}
+}

node-graph/graph-craft/src/application_io/resource/mmap.rs

@@ -0,0 +1,160 @@
+use graphene_application_io::{LoadResource, Resource, ResourceFuture, ResourceHash, ResourceStorage};
+use mmap_io::mmap::{MemoryMappedFile, MmapMode};
+use std::collections::HashMap;
+use std::fs;
+use std::io::Write;
+use std::path::{Path, PathBuf};
+use std::sync::RwLock;
+
+pub struct MmapResourceStorage {
+	root: PathBuf,
+	cache: RwLock<HashMap<ResourceHash, Resource>>,
+}
+
+impl MmapResourceStorage {
+	pub fn new(root: impl Into<PathBuf>) -> std::io::Result<Self> {
+		let root = root.into();
+		fs::create_dir_all(&root)?;
+		Ok(Self {
+			root,
+			cache: RwLock::new(HashMap::new()),
+		})
+	}
+
+	fn path_for(&self, hash: &ResourceHash) -> PathBuf {
+		let hash = String::from(hash);
+		let mut path = self.root.clone();
+		path.push(&hash[..2]);
+		path.push(&hash[2..]);
+		path
+	}
+
+	fn open_mmap(path: &Path) -> Option<MemoryMappedFile> {
+		match MemoryMappedFile::builder(path).mode(MmapMode::ReadOnly).huge_pages(true).open() {
+			Ok(file) => Some(file),
+			Err(error) => {
+				log::warn!("Failed to mmap {path:?} retrying without huge pages: {error}");
+
+				match MemoryMappedFile::open_ro(path) {
+					Ok(file) => Some(file),
+					Err(error) => {
+						log::error!("Failed to mmap {path:?}: {error}");
+						None
+					}
+				}
+			}
+		}
+	}
+
+	fn lookup(&self, hash: &ResourceHash) -> Option<Resource> {
+		if let Some(resource) = self.cache.read().unwrap_or_else(|poisoned| poisoned.into_inner()).get(hash) {
+			return Some(resource.clone());
+		}
+
+		let path = self.path_for(hash);
+		let mmap = Self::open_mmap(&path)?;
+		let resource = Resource::new(MmappedBytes(mmap));
+
+		self.cache.write().unwrap_or_else(|poisoned| poisoned.into_inner()).insert(*hash, resource.clone());
+		Some(resource)
+	}
+}
+
+impl LoadResource for MmapResourceStorage {
+	fn load(&self, hash: ResourceHash) -> ResourceFuture {
+		let result = self.lookup(&hash);
+		Box::pin(async move { result })
+	}
+}
+
+impl ResourceStorage for MmapResourceStorage {
+	fn store(&mut self, data: &[u8]) -> ResourceHash {
+		let hash = ResourceHash::from(data);
+		let path = self.path_for(&hash);
+
+		if path.exists() {
+			return hash;
+		}
+
+		let Some(parent) = path.parent() else {
+			log::error!("Resource path {path:?} has no parent directory");
+			return hash;
+		};
+		if let Err(error) = fs::create_dir_all(parent) {
+			log::error!("Failed to create resource subdirectory {parent:?}: {error}");
+			return hash;
+		}
+
+		let tmp = parent.join(format!(
+			"tmp.{}.{}.{}",
+			path.file_name().and_then(|n| n.to_str()).unwrap_or(""),
+			std::process::id(),
+			std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_nanos()).unwrap_or(0),
+		));
+
+		let write_result = (|| -> std::io::Result<()> {
+			let mut file = fs::OpenOptions::new().write(true).create_new(true).open(&tmp)?;
+			file.write_all(data)?;
+			file.sync_all()?;
+			fs::rename(&tmp, &path)
+		})();
+
+		if let Err(error) = write_result {
+			let _ = fs::remove_file(&tmp);
+			if !path.exists() {
+				log::error!("Failed to write resource to {path:?}: {error}");
+			}
+		}
+
+		hash
+	}
+
+	fn contains(&mut self, hash: &ResourceHash) -> bool {
+		self.cache.get_mut().unwrap_or_else(|poisoned| poisoned.into_inner()).contains_key(hash) || self.path_for(hash).exists()
+	}
+
+	fn garbage_collect(&mut self, used: &[ResourceHash]) {
+		let used_set: std::collections::HashSet<ResourceHash> = used.iter().cloned().collect();
+		self.cache.get_mut().unwrap_or_else(|poisoned| poisoned.into_inner()).retain(|hash, _| used_set.contains(hash));
+
+		let Ok(top_entries) = fs::read_dir(&self.root) else { return };
+		for top_entry in top_entries.flatten() {
+			let top_path = top_entry.path();
+			if !top_path.is_dir() {
+				continue;
+			}
+			let Ok(entries) = fs::read_dir(&top_path) else { continue };
+			for entry in entries.flatten() {
+				let path = entry.path();
+				let Some(prefix) = top_path.file_name().and_then(|n| n.to_str()) else { continue };
+				let Some(suffix) = path.file_name().and_then(|n| n.to_str()) else { continue };
+				if suffix.starts_with("tmp.") {
+					continue;
+				}
+				let hex = format!("{prefix}{suffix}");
+				let Ok(hash) = ResourceHash::try_from(hex.as_str()) else { continue };
+				if !used_set.contains(&hash)
+					&& let Err(error) = fs::remove_file(&path)
+				{
+					log::error!("Failed to remove unused resource {path:?}: {error}");
+				}
+			}
+
+			let _ = fs::remove_dir(&top_path);
+		}
+	}
+}
+
+struct MmappedBytes(MemoryMappedFile);
+impl AsRef<[u8]> for MmappedBytes {
+	fn as_ref(&self) -> &[u8] {
+		let len = self.0.len();
+		match self.0.as_slice(0, len) {
+			Ok(slice) => slice,
+			Err(error) => {
+				log::error!("Failed to obtain mmap slice: {error}");
+				&[]
+			}
+		}
+	}
+}

node-graph/graph-craft/src/application_io/resource/opfs.rs

@@ -0,0 +1,366 @@
+use graphene_application_io::{LoadResource, Resource, ResourceFuture, ResourceHash, ResourceStorage};
+use js_sys::Uint8Array;
+use std::collections::{HashMap, HashSet, VecDeque};
+use std::pin::Pin;
+use std::sync::{Arc, Mutex};
+use std::task::{Context, Poll, Waker};
+use wasm_bindgen::JsCast;
+use wasm_bindgen::prelude::*;
+use wasm_bindgen_futures::{JsFuture, spawn_local};
+use web_sys::{Blob, DomException, FileSystemDirectoryHandle, FileSystemFileHandle, FileSystemGetDirectoryOptions, FileSystemGetFileOptions, FileSystemWritableFileStream, WritableStream};
+
+enum Mutation {
+	Write { hash: ResourceHash, bytes: Arc<[u8]> },
+	Delete { hash: ResourceHash },
+}
+
+struct Inner {
+	directory: FileSystemDirectoryHandle,
+	cache: HashMap<ResourceHash, Resource>,
+	on_disk: HashSet<ResourceHash>,
+	queue: VecDeque<Mutation>,
+	worker_active: bool,
+	persist_requested: bool,
+}
+
+pub struct OpfsResourceStorage {
+	inner: Arc<Mutex<Inner>>,
+}
+
+// SAFETY: This is only compiled for browser wasm, where JS handles remain on the main thread.
+unsafe impl Send for OpfsResourceStorage {}
+unsafe impl Sync for OpfsResourceStorage {}
+
+impl OpfsResourceStorage {
+	pub async fn load(directory_name: &str) -> Result<Self, JsValue> {
+		let directory = open_resource_directory(directory_name).await?;
+		let on_disk = enumerate_hashes(&directory).await?;
+
+		Ok(Self {
+			inner: Arc::new(Mutex::new(Inner {
+				directory,
+				cache: HashMap::new(),
+				on_disk,
+				queue: VecDeque::new(),
+				worker_active: false,
+				persist_requested: false,
+			})),
+		})
+	}
+}
+
+impl LoadResource for OpfsResourceStorage {
+	fn load(&self, hash: ResourceHash) -> ResourceFuture {
+		let inner = self.inner.clone();
+
+		{
+			let guard = inner.lock().unwrap();
+			if let Some(resource) = guard.cache.get(&hash) {
+				let resource = resource.clone();
+				return Box::pin(async move { Some(resource) });
+			}
+			if !guard.on_disk.contains(&hash) {
+				return Box::pin(async move { None });
+			}
+		}
+
+		let (sender, receiver) = oneshot();
+		spawn_local(async move {
+			sender.send(read_from_opfs(inner, hash).await);
+		});
+
+		Box::pin(receiver)
+	}
+}
+
+impl ResourceStorage for OpfsResourceStorage {
+	fn store(&mut self, data: &[u8]) -> ResourceHash {
+		let hash = ResourceHash::from(data);
+		let mut guard = self.inner.lock().unwrap();
+
+		let mut bytes = None;
+		if !guard.cache.contains_key(&hash) {
+			let resource_bytes = Arc::<[u8]>::from(data);
+			guard.cache.insert(hash, Resource::new(resource_bytes.clone()));
+			bytes = Some(resource_bytes);
+		}
+
+		if !guard.on_disk.contains(&hash) {
+			let bytes = bytes.unwrap_or_else(|| Arc::<[u8]>::from(data));
+			guard.on_disk.insert(hash);
+			guard.queue.push_back(Mutation::Write { hash, bytes });
+			kick_worker(&self.inner, &mut guard);
+		}
+
+		hash
+	}
+
+	fn contains(&mut self, hash: &ResourceHash) -> bool {
+		let guard = self.inner.lock().unwrap();
+		guard.cache.contains_key(hash) || guard.on_disk.contains(hash)
+	}
+
+	fn garbage_collect(&mut self, used: &[ResourceHash]) {
+		let used: HashSet<ResourceHash> = used.iter().copied().collect();
+		let mut guard = self.inner.lock().unwrap();
+
+		guard.cache.retain(|hash, _| used.contains(hash));
+
+		let unused: Vec<ResourceHash> = guard.on_disk.iter().copied().filter(|hash| !used.contains(hash)).collect();
+		for hash in unused {
+			guard.on_disk.remove(&hash);
+			guard.queue.push_back(Mutation::Delete { hash });
+		}
+
+		if !guard.queue.is_empty() {
+			kick_worker(&self.inner, &mut guard);
+		}
+	}
+}
+
+fn kick_worker(inner: &Arc<Mutex<Inner>>, guard: &mut Inner) {
+	if guard.worker_active {
+		return;
+	}
+
+	guard.worker_active = true;
+	let inner = inner.clone();
+	spawn_local(drain_queue(inner));
+}
+
+async fn drain_queue(inner: Arc<Mutex<Inner>>) {
+	loop {
+		let (directory, mutation, persist_requested) = {
+			let mut guard = inner.lock().unwrap();
+			let Some(mutation) = guard.queue.pop_front() else {
+				guard.worker_active = false;
+				return;
+			};
+
+			let persist_requested = matches!(mutation, Mutation::Write { .. }) && !guard.persist_requested;
+			if persist_requested {
+				guard.persist_requested = true;
+			}
+
+			(guard.directory.clone(), mutation, persist_requested)
+		};
+
+		if persist_requested {
+			request_persistence().await;
+		}
+
+		match mutation {
+			Mutation::Write { hash, bytes } => {
+				if let Err(error) = write_file(&directory, &hash, &bytes).await {
+					log::error!("OPFS write for {hash} failed: {error:?}");
+				}
+			}
+			Mutation::Delete { hash } => {
+				if let Err(error) = delete_file(&directory, &hash).await {
+					log::error!("OPFS delete for {hash} failed: {error:?}");
+				}
+			}
+		}
+	}
+}
+
+async fn request_persistence() {
+	let Some(window) = web_sys::window() else {
+		log::warn!("OPFS persist() skipped: no window");
+		return;
+	};
+	let storage = window.navigator().storage();
+
+	match storage.persist() {
+		Ok(promise) => match JsFuture::from(promise).await {
+			Ok(value) if value.as_bool() == Some(true) => {}
+			Ok(_) => log::warn!("OPFS persistence was not granted; browser may evict resources under storage pressure"),
+			Err(error) => log::warn!("OPFS persist() rejected: {error:?}"),
+		},
+		Err(error) => log::warn!("OPFS persist() threw: {error:?}"),
+	}
+}
+
+async fn open_resource_directory(directory_name: &str) -> Result<FileSystemDirectoryHandle, JsValue> {
+	let storage = web_sys::window().ok_or_else(|| JsValue::from_str("no window"))?.navigator().storage();
+	let root: FileSystemDirectoryHandle = JsFuture::from(storage.get_directory()).await?.dyn_into()?;
+
+	let options = FileSystemGetDirectoryOptions::new();
+	options.set_create(true);
+
+	JsFuture::from(root.get_directory_handle_with_options(directory_name, &options)).await?.dyn_into()
+}
+
+async fn enumerate_hashes(directory: &FileSystemDirectoryHandle) -> Result<HashSet<ResourceHash>, JsValue> {
+	let iterator = directory.keys();
+	let mut hashes = HashSet::new();
+
+	loop {
+		let next: js_sys::IteratorNext = JsFuture::from(iterator.next()?).await?.unchecked_into();
+		if next.done() {
+			break;
+		}
+
+		let Some(name) = next.value().as_string() else {
+			log::warn!("Skipping non-string OPFS resource entry");
+			continue;
+		};
+
+		match ResourceHash::try_from(name.as_str()) {
+			Ok(hash) => {
+				hashes.insert(hash);
+			}
+			Err(error) => log::warn!("Skipping non-resource OPFS entry {name:?}: {error}"),
+		}
+	}
+
+	Ok(hashes)
+}
+
+async fn write_file(directory: &FileSystemDirectoryHandle, hash: &ResourceHash, bytes: &[u8]) -> Result<(), JsValue> {
+	let options = FileSystemGetFileOptions::new();
+	options.set_create(true);
+
+	let name = file_name(hash);
+	let handle: FileSystemFileHandle = JsFuture::from(directory.get_file_handle_with_options(&name, &options)).await?.dyn_into()?;
+	let writable: FileSystemWritableFileStream = JsFuture::from(handle.create_writable()).await?.dyn_into()?;
+	let stream: WritableStream = writable.clone().unchecked_into();
+	let bytes = Uint8Array::from(bytes);
+
+	if let Err(error) = JsFuture::from(writable.write_with_js_u8_array(&bytes)?).await {
+		let _ = JsFuture::from(stream.abort()).await;
+		return Err(error);
+	}
+
+	JsFuture::from(stream.close()).await?;
+	Ok(())
+}
+
+async fn delete_file(directory: &FileSystemDirectoryHandle, hash: &ResourceHash) -> Result<(), JsValue> {
+	let name = file_name(hash);
+	match JsFuture::from(directory.remove_entry(&name)).await {
+		Ok(_) => Ok(()),
+		Err(error) if is_not_found(&error) => Ok(()),
+		Err(error) => Err(error),
+	}
+}
+
+async fn read_from_opfs(inner: Arc<Mutex<Inner>>, hash: ResourceHash) -> Option<Resource> {
+	let directory = {
+		let guard = inner.lock().unwrap();
+		if let Some(resource) = guard.cache.get(&hash) {
+			return Some(resource.clone());
+		}
+		if !guard.on_disk.contains(&hash) {
+			return None;
+		}
+		guard.directory.clone()
+	};
+
+	let name = file_name(&hash);
+	let handle: FileSystemFileHandle = match JsFuture::from(directory.get_file_handle(&name)).await {
+		Ok(value) => match value.dyn_into() {
+			Ok(handle) => handle,
+			Err(value) => {
+				log::error!("OPFS returned non-file handle for {hash}: {value:?}");
+				return None;
+			}
+		},
+		Err(error) if is_not_found(&error) => return None,
+		Err(error) => {
+			log::error!("OPFS getFileHandle for {hash} failed: {error:?}");
+			return None;
+		}
+	};
+
+	let file = match JsFuture::from(handle.get_file()).await {
+		Ok(value) => value,
+		Err(error) => {
+			log::error!("OPFS getFile for {hash} failed: {error:?}");
+			return None;
+		}
+	};
+	let blob: Blob = match file.dyn_into() {
+		Ok(blob) => blob,
+		Err(value) => {
+			log::error!("OPFS getFile returned non-Blob for {hash}: {value:?}");
+			return None;
+		}
+	};
+	let buffer = match JsFuture::from(blob.array_buffer()).await {
+		Ok(buffer) => buffer,
+		Err(error) => {
+			log::error!("OPFS arrayBuffer for {hash} failed: {error:?}");
+			return None;
+		}
+	};
+
+	let bytes: Arc<[u8]> = Uint8Array::new(&buffer).to_vec().into();
+	let actual = ResourceHash::from(bytes.as_ref());
+	if actual != hash {
+		log::error!("OPFS content-integrity failure: file {hash} hashes to {actual}");
+		return None;
+	}
+
+	let resource = Resource::new(bytes);
+	let mut guard = inner.lock().unwrap();
+	if let Some(resource) = guard.cache.get(&hash) {
+		return Some(resource.clone());
+	}
+	if !guard.on_disk.contains(&hash) {
+		return None;
+	}
+
+	guard.cache.insert(hash, resource.clone());
+	Some(resource)
+}
+
+fn file_name(hash: &ResourceHash) -> String {
+	String::from(hash)
+}
+
+fn is_not_found(error: &JsValue) -> bool {
+	error.dyn_ref::<DomException>().is_some_and(|error| error.name() == "NotFoundError")
+}
+
+fn oneshot() -> (OneshotSender, OneshotReceiver) {
+	let state = Arc::new(Mutex::new(OneshotState::default()));
+	(OneshotSender { state: state.clone() }, OneshotReceiver { state })
+}
+
+#[derive(Default)]
+struct OneshotState {
+	value: Option<Option<Resource>>,
+	waker: Option<Waker>,
+}
+
+struct OneshotSender {
+	state: Arc<Mutex<OneshotState>>,
+}
+
+impl OneshotSender {
+	fn send(self, value: Option<Resource>) {
+		let mut guard = self.state.lock().unwrap();
+		guard.value = Some(value);
+		if let Some(waker) = guard.waker.take() {
+			waker.wake();
+		}
+	}
+}
+
+struct OneshotReceiver {
+	state: Arc<Mutex<OneshotState>>,
+}
+
+impl std::future::Future for OneshotReceiver {
+	type Output = Option<Resource>;
+
+	fn poll(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
+		let mut guard = self.state.lock().unwrap();
+		if let Some(value) = guard.value.take() {
+			return Poll::Ready(value);
+		}
+		guard.waker = Some(context.waker().clone());
+		Poll::Pending
+	}
+}

node-graph/graph-craft/src/application_io/wasm.rs

@@ -1,105 +0,0 @@
-use dyn_any::StaticType;
-use graphene_application_io::{ApplicationError, ApplicationIo, ResourceFuture};
-use std::collections::HashMap;
-use std::sync::Arc;
-#[cfg(feature = "tokio")]
-use tokio::io::AsyncReadExt;
-#[cfg(target_family = "wasm")]
-use wasm_bindgen::JsCast;
-#[cfg(feature = "wgpu")]
-use wgpu_executor::WgpuExecutor;
-
-#[derive(Debug, Default)]
-pub struct WasmApplicationIo {
-	#[cfg(feature = "wgpu")]
-	pub(crate) gpu_executor: Option<WgpuExecutor>,
-	pub resources: HashMap<String, Arc<[u8]>>,
-}
-
-impl WasmApplicationIo {
-	pub async fn new() -> Self {
-		#[cfg(feature = "wgpu")]
-		let executor = if let Some(gpu) = web_sys::window().map(|w| w.navigator().gpu()) {
-			let request_adapter = || {
-				let request_adapter = js_sys::Reflect::get(&gpu, &wasm_bindgen::JsValue::from_str("requestAdapter")).ok()?;
-				let function = request_adapter.dyn_ref::<js_sys::Function>()?;
-				function.call0(&gpu).ok()
-			};
-			let result = request_adapter();
-			match result {
-				None => None,
-				Some(_) => WgpuExecutor::new().await,
-			}
-		} else {
-			None
-		};
-
-		#[cfg(not(feature = "wgpu"))]
-		let wgpu_available = false;
-		#[cfg(feature = "wgpu")]
-		let wgpu_available = executor.is_some();
-		super::set_wgpu_available(wgpu_available);
-
-		let mut io = Self {
-			#[cfg(feature = "wgpu")]
-			gpu_executor: executor,
-			resources: HashMap::new(),
-		};
-		io.resources.insert("null".to_string(), Arc::from(include_bytes!("../null.png").to_vec()));
-
-		io
-	}
-}
-
-impl ApplicationIo for WasmApplicationIo {
-	#[cfg(feature = "wgpu")]
-	type Executor = WgpuExecutor;
-	#[cfg(not(feature = "wgpu"))]
-	type Executor = ();
-
-	#[cfg(feature = "wgpu")]
-	fn gpu_executor(&self) -> Option<&Self::Executor> {
-		self.gpu_executor.as_ref()
-	}
-
-	fn load_resource(&self, url: impl AsRef<str>) -> Result<ResourceFuture, ApplicationError> {
-		let url = url::Url::parse(url.as_ref()).map_err(|_| ApplicationError::InvalidUrl)?;
-		log::trace!("Loading resource: {url:?}");
-		match url.scheme() {
-			#[cfg(feature = "tokio")]
-			"file" => {
-				let path = url.to_file_path().map_err(|_| ApplicationError::NotFound)?;
-				let path = path.to_str().ok_or(ApplicationError::NotFound)?;
-				let path = path.to_owned();
-				Ok(Box::pin(async move {
-					let file = tokio::fs::File::open(path).await.map_err(|_| ApplicationError::NotFound)?;
-					let mut reader = tokio::io::BufReader::new(file);
-					let mut data = Vec::new();
-					reader.read_to_end(&mut data).await.map_err(|_| ApplicationError::NotFound)?;
-					Ok(Arc::from(data))
-				}) as ResourceFuture)
-			}
-			"http" | "https" => {
-				let url = url.to_string();
-				Ok(Box::pin(async move {
-					let client = reqwest::Client::new();
-					let response = client.get(url).send().await.map_err(|_| ApplicationError::NotFound)?;
-					let data = response.bytes().await.map_err(|_| ApplicationError::NotFound)?;
-					Ok(Arc::from(data.to_vec()))
-				}) as ResourceFuture)
-			}
-			"graphite" => {
-				let path = url.path();
-				let path = path.to_owned();
-				log::trace!("Loading local resource: {path}");
-				let data = self.resources.get(&path).ok_or(ApplicationError::NotFound)?.clone();
-				Ok(Box::pin(async move { Ok(data.clone()) }) as ResourceFuture)
-			}
-			_ => Err(ApplicationError::NotFound),
-		}
-	}
-}
-
-unsafe impl StaticType for WasmApplicationIo {
-	type Static = WasmApplicationIo;
-}

node-graph/graph-craft/src/document/value.rs

@@ -396,6 +396,7 @@ tagged_value! {
 	Footprint(Footprint),
 	VectorModification(Box<VectorModification>),
 	ImageData(Image<Color>),
+	Resource(graphene_application_io::ResourceHash),
 	// ==========
 	// ENUM TYPES
 	// ==========