add wasm compatibility

Author: arriqaaq

Cargo.lock

@@ -39,7 +39,6 @@ guardian = "1.3.0"
 scopeguard = "1.2.0"
 memmap2 = "0.9.5"
 interval-heap = "0.0.5"
-thiserror = "1.0"
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 fs2 = "0.4.3"

Cargo.toml

@@ -55,7 +55,6 @@ let tree = TreeBuilder::new()
     .with_max_memtable_size(100 * 1024 * 1024)         // 100MB memtable size
     .with_block_size(4096)                             // 4KB block size
     .with_level_count(1)                               // Number of levels in LSM tree
-    .with_vlog_value_threshold(4096)                   // Values smaller than this stored inline
     .with_vlog_max_file_size(128 * 1024 * 1024)        // 128MB VLog file size
     .with_enable_vlog(true)                            // Enable/disable VLog
     .build()?
@@ -70,7 +69,6 @@ let tree = TreeBuilder::new()
 
 ### VLog Options
 
-- `vlog_value_threshold`: Values smaller than this are stored inline in SSTables
 - `vlog_max_file_size`: Maximum size of VLog files before rotation
 - `with_enable_vlog()`: Enable/disable Value Log for large value storage
 - `vlog_gc_discard_ratio`: Threshold for triggering VLog garbage collection

README.md

@@ -628,7 +628,8 @@ mod tests {
 		// Now they should encode identically (both write value_len=0)
 		assert_eq!(encoded_none, encoded_empty, "None and Some(&[]) should now encode identically");
 
-		// Test reading them back - both should return None (since both encode as value_len=0)
+		// Test reading them back - both should return None (since both encode as
+		// value_len=0)
 		let mut reader_none = BatchReader::new(&encoded_none).unwrap();
 		let (kind1, key1, value1) = reader_none.read_record().unwrap().unwrap();
 		assert_eq!(kind1, InternalKeyKind::Delete);

src/batch.rs

@@ -1,9 +1,10 @@
-use quick_cache::Weighter;
-use quick_cache::{sync::Cache as QCache, Equivalent};
+use quick_cache::sync::Cache as QCache;
+use quick_cache::{Equivalent, Weighter};
 use std::sync::atomic::AtomicU64;
 use std::sync::Arc;
 
-use crate::sstable::{block::Block, InternalKeyTrait};
+use crate::sstable::block::Block;
+use crate::sstable::InternalKeyTrait;
 use crate::Value;
 
 pub type CacheID = u64;

src/cache.rs

@@ -67,7 +67,8 @@ impl CheckpointMetadata {
 		}
 	}
 
-	/// Checks if this metadata version is compatible with current implementation
+	/// Checks if this metadata version is compatible with current
+	/// implementation
 	pub fn is_compatible(&self) -> bool {
 		// For now, we only support version 1
 		self.version == CHECKPOINT_VERSION
@@ -259,8 +260,9 @@ impl<K: InternalKeyTrait> DatabaseCheckpoint<K> {
 
 	/// Flushes all memtables to ensure checkpoint consistency
 	fn flush_all_memtables(&self) -> Result<()> {
-		// Keep calling compact_memtable until all memtables (active + immutable) are flushed
-		// compact_memtable already handles the logic of checking if there's anything to flush
+		// Keep calling compact_memtable until all memtables (active + immutable) are
+		// flushed compact_memtable already handles the logic of checking if there's
+		// anything to flush
 		loop {
 			// Check if there are any memtables to flush
 			let has_active = {
@@ -381,7 +383,8 @@ impl<K: InternalKeyTrait> DatabaseCheckpoint<K> {
 		CheckpointMetadata::from_bytes(&data)
 	}
 
-	/// Helper function to copy a directory recursively (synchronous version to avoid recursion issues)
+	/// Helper function to copy a directory recursively (synchronous version to
+	/// avoid recursion issues)
 	fn copy_directory_sync(source: &Path, dest: &Path) -> Result<u64> {
 		if !source.exists() {
 			return Ok(0);
@@ -484,7 +487,8 @@ mod tests {
 		let bytes = metadata.to_bytes().unwrap();
 
 		// Verify the binary format structure
-		// 4 bytes version + 8 bytes timestamp + 8 bytes seq + 8 bytes count + 8 bytes size = 36 bytes
+		// 4 bytes version + 8 bytes timestamp + 8 bytes seq + 8 bytes count + 8 bytes
+		// size = 36 bytes
 		assert_eq!(bytes.len(), 36);
 
 		// Check that version is at the beginning (big endian)

src/checkpoint.rs

@@ -1,16 +1,12 @@
 // This commit pipeline is inspired by Pebble's commit pipeline.
 
-use std::sync::{
-	atomic::{AtomicBool, AtomicPtr, AtomicU64, Ordering},
-	Arc, Mutex,
-};
+use std::sync::atomic::{AtomicBool, AtomicPtr, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
 
 use tokio::sync::{oneshot, Semaphore};
 
-use crate::{
-	batch::Batch,
-	error::{Error, Result},
-};
+use crate::batch::Batch;
+use crate::error::{Error, Result};
 
 const MAX_CONCURRENT_COMMITS: usize = 16;
 const DEQUEUE_BITS: u32 = 32;
@@ -132,7 +128,8 @@ impl CommitQueue {
 		self.head_tail.fetch_add(1 << DEQUEUE_BITS, Ordering::Release);
 	}
 
-	// Multi-consumer dequeue - removes the earliest enqueued Batch, if it is applied
+	// Multi-consumer dequeue - removes the earliest enqueued Batch, if it is
+	// applied
 	fn dequeue_applied(&self) -> Option<Arc<CommitBatch>> {
 		loop {
 			let ptrs = self.head_tail.load(Ordering::Acquire);
@@ -210,8 +207,8 @@ impl CommitPipeline {
 	}
 
 	/// Synchronous commit that bypasses the async pipeline machinery.
-	/// This is suitable for cases where no flow control or async coordination is needed,
-	/// such as delete list updates during compaction.
+	/// This is suitable for cases where no flow control or async coordination
+	/// is needed, such as delete list updates during compaction.
 	pub(crate) fn sync_commit(&self, batch: Batch, sync_wal: bool) -> Result<()> {
 		if self.shutdown.load(Ordering::Acquire) {
 			return Err(Error::PipelineStall);
@@ -372,7 +369,7 @@ mod tests {
 	use std::time::Duration;
 
 	use crate::sstable::InternalKeyKind;
-	use crate::spawn::spawn;
+	use crate::util::spawn;
 
 	use super::*;
 
@@ -435,11 +432,14 @@ mod tests {
 	#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
 	async fn test_concurrent_commits() {
 		let pipeline = CommitPipeline::new(Arc::new(MockEnv));
+		let num_tasks = 10;
+		let (result_sender, result_receiver) = async_channel::bounded(num_tasks);
 
-		let mut handles = vec![];
-		for i in 0..10 {
+		// Spawn concurrent commit tasks
+		for i in 0..num_tasks {
 			let pipeline = pipeline.clone();
-			let handle = spawn(async move {
+			let sender = result_sender.clone();
+			spawn(async move {
 				let mut batch = Batch::new();
 				batch
 					.add_record(
@@ -448,13 +448,20 @@ mod tests {
 						Some(&[1, 2, 3]),
 					)
 					.unwrap();
-				pipeline.commit(batch, false).await
+				let result = pipeline.commit(batch, false).await;
+				let _ = sender.send((i, result)).await;
 			});
-			handles.push(handle);
 		}
 
-		for (i, handle) in handles.into_iter().enumerate() {
-			let result = handle.await.unwrap();
+		// Collect all results
+		let mut results = Vec::new();
+		for _ in 0..num_tasks {
+			let (i, result) = result_receiver.recv().await.unwrap();
+			results.push((i, result));
+		}
+
+		// Verify all commits succeeded
+		for (i, result) in results {
 			assert!(result.is_ok(), "Commit {i} failed: {result:?}");
 		}
 
@@ -494,11 +501,14 @@ mod tests {
 	#[tokio::test(flavor = "multi_thread")]
 	async fn test_concurrent_commits_with_delays() {
 		let pipeline = CommitPipeline::new(Arc::new(DelayedMockEnv));
+		let num_tasks = 5;
+		let (result_sender, result_receiver) = async_channel::bounded(5);
 
-		let mut handles = vec![];
-		for i in 0..5 {
+		// Spawn concurrent commit tasks
+		for i in 0..num_tasks {
 			let pipeline = pipeline.clone();
-			let handle = spawn(async move {
+			let sender = result_sender.clone();
+			spawn(async move {
 				let mut batch = Batch::new();
 				batch
 					.add_record(
@@ -507,17 +517,19 @@ mod tests {
 						Some(&[1, 2, 3]),
 					)
 					.unwrap();
-				pipeline.commit(batch, false).await
+				let result = pipeline.commit(batch, false).await;
+				let _ = sender.send(result).await;
 			});
-			handles.push(handle);
 		}
 
-		for handle in handles {
-			assert!(handle.await.unwrap().is_ok());
+		// Collect all results
+		for _ in 0..num_tasks {
+			let res = result_receiver.recv().await.unwrap();
+			assert!(res.is_ok());
 		}
 
 		// Verify sequence numbers are published correctly
-		assert_eq!(pipeline.get_visible_seq_num(), 5);
+		assert_eq!(pipeline.get_visible_seq_num(), num_tasks);
 
 		// Shutdown the pipeline
 		pipeline.shutdown();

src/commit.rs

@@ -1,25 +1,19 @@
-use crate::{
-	compaction::{CompactionChoice, CompactionInput, CompactionStrategy},
-	error::Result,
-	iter::{BoxedIterator, CompactionIterator},
-	levels::{write_manifest_to_disk, LevelManifest, ManifestChangeSet},
-	lsm::CoreInner,
-	memtable::ImmutableMemtables,
-	sstable::{
-		table::{Table, TableWriter},
-		InternalKeyTrait,
-	},
-	vfs::File,
-	vlog::VLog,
-	Options as LSMOptions,
-};
-
-use std::{
-	collections::HashMap,
-	fs::File as SysFile,
-	path::{Path, PathBuf},
-	sync::{Arc, RwLock, RwLockWriteGuard},
-};
+use crate::compaction::{CompactionChoice, CompactionInput, CompactionStrategy};
+use crate::error::Result;
+use crate::iter::{BoxedIterator, CompactionIterator};
+use crate::levels::{write_manifest_to_disk, LevelManifest, ManifestChangeSet};
+use crate::lsm::CoreInner;
+use crate::memtable::ImmutableMemtables;
+use crate::sstable::table::{Table, TableWriter};
+use crate::sstable::InternalKeyTrait;
+use crate::vfs::File;
+use crate::vlog::VLog;
+use crate::Options as LSMOptions;
+
+use std::collections::HashMap;
+use std::fs::File as SysFile;
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, RwLock, RwLockWriteGuard};
 
 /// Compaction options
 pub(crate) struct CompactionOptions<K: InternalKeyTrait> {
@@ -175,7 +169,8 @@ impl<K: InternalKeyTrait> Compactor<K> {
 		// Create a changeset for the compaction
 		let mut changeset = ManifestChangeSet::<K>::default();
 
-		// Add tables to delete (the ones being merged) - use the same efficient approach as original
+		// Add tables to delete (the ones being merged) - use the same efficient
+		// approach as original
 		for (level_idx, level) in manifest.levels.get_levels().iter().enumerate() {
 			for &table_id in &input.tables_to_merge {
 				if level.tables.iter().any(|t| t.id == table_id) {