Update comments

Author: Goshawk

Cargo.toml

@@ -6,7 +6,6 @@ edition = "2021"
 [dependencies]
 tokio = { version = "1", features = ["rt", "net", "sync", "time", "io-std", "rt-multi-thread", "macros"] }
 thiserror = "1.0.63"
-async-trait = "0.1.81"
 requestty = "0.5.0"
 hex = "0.4.3"
 aws-config = "1.5.4"

README.md

@@ -6,7 +6,7 @@ Data lake implementation integrated with AWS S3
 - Async-Download chunks from AWS S3
 - Persist on-disk in a lock-less manner
 - List all persisted chunks by ID from a cache
-- Find and lock a chunk - Once locked, chunk cannot be deleted
+- Find and lock a chunk - Once locked, chunk cannot be deleted until the DataChunkRef is released/dropped.
 - Scheduled deletion - Scheduled for deletion, a chunk will be removed once it is no longer in use.
 
 - Maximum allocated on-disk storage limit

src/data_manager.rs

@@ -3,7 +3,7 @@ use crate::{
 };
 use aws_config::BehaviorVersion;
 use serde_binary::binary_stream::Endian;
-use std::collections::HashMap;
+use std::collections::{hash_map, HashMap};
 use std::path::PathBuf;
 use std::sync::Arc;
 use tokio::sync::{
@@ -199,7 +199,6 @@ impl<T: StorageEngine> DataManagerImpl<T> {
             .await
             .expect("valid data blob");
 
-           
         let chunk: DataChunk =
             serde_binary::from_vec(vec, Endian::Big)
                 .expect("valid chunk");
@@ -208,27 +207,28 @@ impl<T: StorageEngine> DataManagerImpl<T> {
         // Check if the chunk already exists in the cache
         let exists = {
             let mut cache = cache.write().await;
-            if !cache.contains_key(&chunk.id) {
-                cache.insert(
-                    chunk.id,
-                    Arc::new(Semaphore::new(1)),
-                );
+            if let hash_map::Entry::Vacant(e) =
+                cache.entry(chunk.id)
+            {
+                e.insert(Arc::new(Semaphore::new(1)));
                 false
             } else {
                 true
-            }    
+            }
         };
 
         if !exists {
             // Due to the usage RocksDB::OptimisticTransaction it is safe here to use read lock
-            // Any write conflict will be resolved by the RocksDB::commit itself
+            // Any write conflict will be detected by the RocksDB::commit itself.
+            // However, we don't expect high lock contention here.
             if db
                 .read()
                 .await
                 .persist_chunk(chunk, chunk_size)
                 .is_err()
             {
-                // chunk could not be persisted. Rollback the cache
+                // chunk could not be persisted.
+                // Rollback it from the cache to keep the cache consistent
                 cache.write().await.remove(&id);
             }
         }

src/lib.rs

@@ -1,10 +1,10 @@
 pub mod data_manager;
 pub mod rocksdb_storage_engine;
 
+use serde_binary::Encode;
 use std::collections::HashMap;
 use std::ops::Range;
 use std::path::{Path, PathBuf};
-use serde_binary::Encode;
 use thiserror::Error;
 
 pub type DatasetId = [u8; 32];
@@ -27,7 +27,10 @@ pub struct DataChunk {
 }
 
 impl Encode for DataChunk {
-    fn encode(&self, _serializer: &mut serde_binary::Serializer<'_>) -> Result<(), serde_binary::Error> {
+    fn encode(
+        &self,
+        _serializer: &mut serde_binary::Serializer<'_>,
+    ) -> Result<(), serde_binary::Error> {
         // TODO: Implement
         Ok(())
     }
@@ -47,7 +50,6 @@ pub enum Error {
     MaxSizeAllocated(u32),
 }
 
-// Implement From To rocksdb_lib::Error for Error
 impl From<rocksdb_lib::Error> for Error {
     fn from(e: rocksdb_lib::Error) -> Self {
         Self::InternalErr(e)

src/rocksdb_storage_engine.rs

@@ -12,34 +12,12 @@ use serde_binary::binary_stream::Endian;
 
 const SIZE_KEY: [u8; 1] = [1u8; 1];
 
+/// Represents a storage engine implementation using RocksDB.
 pub struct StorageEngineImpl {
     rocksdb: OptimisticTransactionDB,
     conf: StorageConf,
 }
 
-/// Implementation of storage engine for a data lake.
-///
-/// This storage engine provides methods for managing data chunks in a data lake.
-/// It uses an underlying RocksDB database for storing and retrieving data.
-///
-/// # Examples
-///
-/// ```
-/// use datalake::StorageEngine;
-///
-/// // Create a new storage engine
-/// let conf = StorageEngine::new("/path/to/database");
-/// let storage_engine = StorageEngine::from_conf(conf);
-///
-/// // Persist a data chunk
-/// let chunk = DataChunk::new(...);
-/// storage_engine.persist_chunk(&chunk, 100).unwrap();
-///
-/// // Find a chunk ID
-/// let dataset_id = DatasetId::new(...);
-/// let block_number = 42;
-/// let chunk_id = storage_engine.find_chunk_id(dataset_id, block_number).unwrap();
-/// ```
 impl StorageEngine for StorageEngineImpl {
     fn from_conf(conf: StorageConf) -> Self {
         let rocksdb =
@@ -137,8 +115,9 @@ impl StorageEngine for StorageEngineImpl {
             // Persist KEY - VALUE
             // Chunk_ID -> SizeU32 + Chunk Bytes
             let id = chunk.id;
-            let chunk_bytes = serde_binary::encode(&chunk, Endian::Big)
-            .expect("should encode");
+            let chunk_bytes =
+                serde_binary::encode(&chunk, Endian::Big)
+                    .expect("should encode");
 
             // Chunk is now in form of chunk_bytes, drop the chunk struct to free memory
             drop(chunk);