V2

Author: timon-schelling

node-graph/graph-craft/Cargo.toml

@@ -66,6 +66,7 @@ web-sys = { workspace = true, features = [
 	"DomStringList",
 	"Event",
 	"EventTarget",
+	"Window",
 	"StorageManager",
 	"FileSystemDirectoryHandle",
 	"FileSystemFileHandle",

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

@@ -7,7 +7,7 @@ use std::task::{Context, Poll, Waker};
 use wasm_bindgen::JsCast;
 use wasm_bindgen::prelude::*;
 use wasm_bindgen_futures::{JsFuture, spawn_local};
-use web_sys::{DomException, FileSystemDirectoryHandle, FileSystemFileHandle, FileSystemGetDirectoryOptions, FileSystemGetFileOptions, FileSystemWritableFileStream};
+use web_sys::{Blob, DomException, FileSystemDirectoryHandle, FileSystemFileHandle, FileSystemGetDirectoryOptions, FileSystemGetFileOptions, FileSystemWritableFileStream, WritableStream};
 
 enum Mutation {
 	Write { hash: ResourceHash, bytes: Arc<[u8]> },
@@ -32,16 +32,15 @@ unsafe impl Sync for OpfsResourceStorage {}
 
 impl OpfsResourceStorage {
 	pub async fn load(directory_name: &str) -> Result<Self, JsValue> {
-		let root = open_root_directory(directory_name).await?;
-
-		let on_disk = read_all_keys(&root).await.unwrap_or_else(|error| {
+		let directory = open_resource_directory(directory_name).await?;
+		let on_disk = enumerate_hashes(&directory).await.unwrap_or_else(|error| {
 			log::error!("Failed to enumerate existing OPFS resource keys: {error:?}");
 			HashSet::new()
 		});
 
 		Ok(Self {
 			inner: Arc::new(Mutex::new(Inner {
-				directory: root,
+				directory,
 				cache: HashMap::new(),
 				on_disk,
 				queue: VecDeque::new(),
@@ -68,8 +67,7 @@ impl Resources for OpfsResourceStorage {
 
 		let (sender, receiver) = oneshot();
 		spawn_local(async move {
-			let result = fetch_and_verify(inner, hash).await;
-			sender.send(result);
+			sender.send(read_from_opfs(inner, hash).await);
 		});
 
 		Box::pin(receiver)
@@ -83,15 +81,17 @@ impl ResourceStorage for OpfsResourceStorage {
 
 	fn write(&mut self, data: &[u8]) -> ResourceHash {
 		let hash = ResourceHash::from(data);
-		let bytes: Arc<[u8]> = Arc::from(data);
-		let resource = Resource::new(bytes.clone());
-
 		let mut guard = self.inner.lock().unwrap();
-		guard.cache.insert(hash, resource);
 
-		let needs_persist = !guard.on_disk.contains(&hash) || queue_has_delete_for(&guard.queue, &hash);
+		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 needs_persist {
+		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);
@@ -106,57 +106,48 @@ impl ResourceStorage for OpfsResourceStorage {
 	}
 
 	fn garbage_collect(&mut self, used: &[ResourceHash]) {
-		let used_set: HashSet<ResourceHash> = used.iter().copied().collect();
+		let used: HashSet<ResourceHash> = used.iter().copied().collect();
 		let mut guard = self.inner.lock().unwrap();
 
-		guard.cache.retain(|hash, _| used_set.contains(hash));
+		guard.cache.retain(|hash, _| used.contains(hash));
 
-		let to_delete: Vec<ResourceHash> = guard.on_disk.iter().copied().filter(|hash| !used_set.contains(hash)).collect();
-
-		if to_delete.is_empty() {
-			return;
+		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 });
 		}
 
-		for hash in &to_delete {
-			guard.on_disk.remove(hash);
-			guard.queue.push_back(Mutation::Delete { hash: *hash });
+		if !guard.queue.is_empty() {
+			kick_worker(&self.inner, &mut guard);
 		}
-
-		kick_worker(&self.inner, &mut guard);
 	}
 }
 
-fn queue_has_delete_for(queue: &VecDeque<Mutation>, hash: &ResourceHash) -> bool {
-	queue.iter().any(|mutation| matches!(mutation, Mutation::Delete { hash: other } if other == hash))
-}
-
 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 (op, directory) = {
+		let (directory, mutation) = {
 			let mut guard = inner.lock().unwrap();
-			match guard.queue.pop_front() {
-				Some(op) => (op, guard.directory.clone()),
-				None => {
-					guard.worker_active = false;
-					return;
-				}
-			}
+			let Some(mutation) = guard.queue.pop_front() else {
+				guard.worker_active = false;
+				return;
+			};
+			(guard.directory.clone(), mutation)
 		};
 
-		match op {
+		match mutation {
 			Mutation::Write { hash, bytes } => {
 				if let Err(error) = write_file(&directory, &hash, &bytes).await {
 					log::error!("OPFS write for {hash} failed: {error:?}");
-					inner.lock().unwrap().on_disk.remove(&hash);
 				}
 			}
 			Mutation::Delete { hash } => {
@@ -168,103 +159,94 @@ async fn drain_queue(inner: Arc<Mutex<Inner>>) {
 	}
 }
 
-async fn open_root_directory(directory_name: &str) -> Result<FileSystemDirectoryHandle, JsValue> {
+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 opfs_root: FileSystemDirectoryHandle = JsFuture::from(storage.get_directory()).await?.dyn_into()?;
-
-	let create = FileSystemGetDirectoryOptions::new();
-	create.set_create(true);
+	let options = FileSystemGetDirectoryOptions::new();
+	options.set_create(true);
 
-	JsFuture::from(opfs_root.get_directory_handle_with_options(directory_name, &create)).await?.dyn_into()
+	JsFuture::from(root.get_directory_handle_with_options(directory_name, &options)).await?.dyn_into()
 }
 
-async fn read_all_keys(directory: &FileSystemDirectoryHandle) -> Result<HashSet<ResourceHash>, JsValue> {
+async fn enumerate_hashes(directory: &FileSystemDirectoryHandle) -> Result<HashSet<ResourceHash>, JsValue> {
 	let iterator = directory.keys();
-	let mut keys = HashSet::new();
+	let mut hashes = HashSet::new();
 
 	loop {
-		let next: js_sys::IteratorNext = JsFuture::from(iterator.next_iterator()?).await?.unchecked_into();
+		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 entry name");
+			log::warn!("Skipping non-string OPFS resource entry");
 			continue;
 		};
 
 		match ResourceHash::try_from(name.as_str()) {
 			Ok(hash) => {
-				keys.insert(hash);
+				hashes.insert(hash);
 			}
-			Err(error) => log::warn!("Skipping unparseable OPFS entry {name:?}: {error}"),
+			Err(error) => log::warn!("Skipping non-resource OPFS entry {name:?}: {error}"),
 		}
 	}
 
-	Ok(keys)
-}
-
-fn file_name(hash: &ResourceHash) -> String {
-	String::from(hash)
+	Ok(hashes)
 }
 
 async fn write_file(directory: &FileSystemDirectoryHandle, hash: &ResourceHash, bytes: &[u8]) -> Result<(), JsValue> {
-	let create = FileSystemGetFileOptions::new();
-	create.set_create(true);
+	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, &create)).await?.dyn_into()?;
-
+	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);
 
-	let view = Uint8Array::new_with_length(bytes.len() as u32);
-	view.copy_from(bytes);
-
-	let write_result = JsFuture::from(writable.write_with_js_u8_array(&view)?).await;
-	let close_result = JsFuture::from(writable.close()).await;
+	if let Err(error) = JsFuture::from(writable.write_with_js_u8_array(&bytes)?).await {
+		let _ = JsFuture::from(stream.abort()).await;
+		return Err(error);
+	}
 
-	write_result?;
-	close_result?;
+	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(())
-			} else {
-				Err(error)
-			}
-		}
+		Err(error) if is_not_found(&error) => Ok(()),
+		Err(error) => Err(error),
 	}
 }
 
-fn is_not_found(error: &JsValue) -> bool {
-	error.dyn_ref::<DomException>().is_some_and(|error| error.name() == "NotFoundError")
-}
+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()
+	};
 
-async fn fetch_and_verify(inner: Arc<Mutex<Inner>>, hash: ResourceHash) -> Option<Resource> {
-	let directory = inner.lock().unwrap().directory.clone();
 	let name = file_name(&hash);
-
-	let handle = match JsFuture::from(directory.get_file_handle(&name)).await {
-		Ok(value) => match value.dyn_into::<FileSystemFileHandle>() {
+	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 get_file_handle returned non-FileSystemFileHandle for {hash}: {value:?}");
+				log::error!("OPFS returned non-file handle for {hash}: {value:?}");
 				return None;
 			}
 		},
+		Err(error) if is_not_found(&error) => return None,
 		Err(error) => {
-			if is_not_found(&error) {
-				inner.lock().unwrap().on_disk.remove(&hash);
-				return None;
-			}
-			log::error!("OPFS get_file_handle for {hash} failed: {error:?}");
+			log::error!("OPFS getFileHandle for {hash} failed: {error:?}");
 			return None;
 		}
 	};
@@ -276,39 +258,49 @@ async fn fetch_and_verify(inner: Arc<Mutex<Inner>>, hash: ResourceHash) -> Optio
 			return None;
 		}
 	};
-	let blob: web_sys::Blob = match file.dyn_into() {
+	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!("Reading OPFS array buffer for {hash} failed: {error:?}");
+			log::error!("OPFS arrayBuffer for {hash} failed: {error:?}");
 			return None;
 		}
 	};
-	let array = Uint8Array::new(&buffer);
-	let bytes: Arc<[u8]> = Arc::from(array.to_vec());
 
+	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}, removing");
-		let mut guard = inner.lock().unwrap();
-		guard.on_disk.remove(&hash);
-		guard.queue.push_back(Mutation::Delete { hash });
-		kick_worker(&inner, &mut guard);
+		log::error!("OPFS content-integrity failure: file {hash} hashes to {actual}");
 		return None;
 	}
 
 	let resource = Resource::new(bytes);
-	inner.lock().unwrap().cache.insert(hash, resource.clone());
+	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 })