/* This file is part of DarkFi (https://dark.fi) * * Copyright (C) 2020-2025 Dyne.org foundation * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU Affero General Public License as * published by the Free Software Foundation, either version 3 of the * License, or (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU Affero General Public License for more details. * * You should have received a copy of the GNU Affero General Public License * along with this program. If not, see . */ //! Chunk-based file storage implementation. //! This is a building block for a DHT or something similar. //! //! The API supports file insertion and retrieval. There is intentionally no //! `remove` support. File removal should be handled externally, and then it //! is only required to run `garbage_collect()` to clean things up. //! //! The filesystem hierarchy stores two directories: `files` and `chunks`. //! `chunks` store [`MAX_CHUNK_SIZE`] files, where the filename is a BLAKE3 //! hash of the chunk's contents. //! `files` store metadata about a full file, which can be retrieved by //! concatenating the chunks in order. The filename of a file in `files` //! is the BLAKE3 hash of hashed chunks in the correct order. //! //! It might look like the following: //! ``` //! /files/B9fFKaEYphw2oH5PDbeL1TTAcSzL6ax84p8SjBKzuYzX //! /files/... //! /chunks/2bQPxSR8Frz7S7JW3DRAzEtkrHfLXB1CN65V7az77pUp //! /chunks/CvjvN6MfWQYK54DgKNR7MPgFSZqsCgpWKF2p8ot66CCP //! /chunks/... //! ``` //! //! In the above example, contents of `B9fFKaEYphw2oH5PDbeL1TTAcSzL6ax84p8SjBKzuYzX` //! may be: //! ``` //! 2bQPxSR8Frz7S7JW3DRAzEtkrHfLXB1CN65V7az77pUp //! CvjvN6MfWQYK54DgKNR7MPgFSZqsCgpWKF2p8ot66CCP //! ``` //! //! This means, in order to retrieve `B9fFKaEYphw2oH5PDbeL1TTAcSzL6ax84p8SjBKzuYzX`, //! we need to concatenate the files under `/chunks` whose filenames are the //! hashes found above. The contents of the files in `/chunks` are arbitrary //! data, and by concatenating them we can retrieve the original file. //! //! It is important to note that multiple files can use the same chunks. //! This is some kind of naive deduplication, so we actually don't consider //! chunks to be specific to a single file and therefore when we do garbage //! collection, we keep chunks and files independent of each other. use std::{collections::HashSet, path::PathBuf}; use futures::AsyncRead; use log::{debug, info, warn}; use smol::{ fs::{self, File, OpenOptions}, io::{ self, AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader, Cursor, SeekFrom, }, stream::StreamExt, }; use crate::{Error, Result}; /// Defined maximum size of a stored chunk (256 KiB) pub const MAX_CHUNK_SIZE: usize = 262_144; /// Path prefix where file metadata is stored const FILES_PATH: &str = "files"; /// Path prefix where file chunks are stored const CHUNKS_PATH: &str = "chunks"; pub fn hash_to_string(hash: &blake3::Hash) -> String { bs58::encode(hash.as_bytes()).into_string() } /// `ChunkedFile` is a representation of a file we're trying to /// retrieve from `Geode`. /// /// The tuple contains `blake3::Hash` of /// the file's chunks and an optional `PathBuf` which points to /// the filesystem where the chunk can be found. If `None`, it /// is to be assumed that the chunk is not available locally. #[derive(Clone)] pub struct ChunkedFile(Vec<(blake3::Hash, Option)>); impl ChunkedFile { fn new(hashes: &[blake3::Hash]) -> Self { Self(hashes.iter().map(|x| (*x, None)).collect()) } /// Check whether we have all the chunks available locally. pub fn is_complete(&self) -> bool { !self.0.iter().any(|(_, p)| p.is_none()) } /// Return an iterator over the chunks and their paths. pub fn iter(&self) -> core::slice::Iter<'_, (blake3::Hash, Option)> { self.0.iter() } /// Return the number of chunks. pub fn len(&self) -> usize { self.0.len() } /// Return `true` if the chunked file contains no chunk. pub fn is_empty(&self) -> bool { self.0.is_empty() } /// Return the number of chunks available locally. pub fn local_chunks(&self) -> usize { self.0.iter().filter(|(_, p)| p.is_some()).count() } } /// Chunk-based file storage interface. pub struct Geode { /// Path to the filesystem directory where file metadata is stored files_path: PathBuf, /// Path to the filesystem directory where file chunks are stored chunks_path: PathBuf, } /// smol::fs::File::read does not guarantee that the buffer will be filled, even if the buffer is /// smaller than the file. This is a workaround. /// This reads the stream until the buffer is full or until we reached the end of the stream. pub async fn read_until_filled( mut stream: impl AsyncRead + Unpin, buffer: &mut [u8], ) -> io::Result { let mut total_bytes_read = 0; while total_bytes_read < buffer.len() { let bytes_read = stream.read(&mut buffer[total_bytes_read..]).await?; if bytes_read == 0 { break; // EOF reached } total_bytes_read += bytes_read; } Ok(total_bytes_read) } impl Geode { /// Instantiate a new [`Geode`] object. /// `base_path` defines the root directory where Geode will store its /// file metadata and chunks. pub async fn new(base_path: &PathBuf) -> Result { let mut files_path: PathBuf = base_path.into(); let mut chunks_path: PathBuf = base_path.into(); files_path.push(FILES_PATH); chunks_path.push(CHUNKS_PATH); // Create necessary directory structure if needed fs::create_dir_all(&files_path).await?; fs::create_dir_all(&chunks_path).await?; Ok(Self { files_path, chunks_path }) } /// Attempt to read chunk hashes from a given file path and return /// a `Vec` containing the hashes in order. async fn read_metadata(path: &PathBuf) -> Result> { debug!(target: "geode::read_metadata()", "Reading chunks from {:?}", path); let fd = File::open(path).await?; let mut read_chunks = vec![]; let mut lines = BufReader::new(fd).lines(); while let Some(line) = lines.next().await { let line = line?; let mut hash_buf = [0u8; 32]; bs58::decode(line).onto(&mut hash_buf)?; let chunk_hash = blake3::Hash::from_bytes(hash_buf); read_chunks.push(chunk_hash); } Ok(read_chunks) } /// Perform garbage collection over the filesystem hierarchy. /// Returns sets representing deleted files and deleted chunks, respectively. pub async fn garbage_collect(&self) -> Result<(HashSet, HashSet)> { info!(target: "geode::garbage_collect()", "[Geode] Performing garbage collection"); // We track corrupt files and chunks here. let mut deleted_files = HashSet::new(); let mut deleted_chunks = HashSet::new(); let mut deleted_chunk_paths = HashSet::new(); // Scan through available chunks and check them for consistency. let mut chunk_paths = fs::read_dir(&self.chunks_path).await?; let mut buf = [0u8; MAX_CHUNK_SIZE]; while let Some(chunk) = chunk_paths.next().await { let Ok(entry) = chunk else { continue }; let chunk_path = entry.path(); // Skip if we're not a plain file if !chunk_path.is_file() { continue } // Make sure that the filename is a BLAKE3 hash let file_name = match chunk_path.file_name().and_then(|n| n.to_str()) { Some(v) => v, None => continue, }; let mut hash_buf = [0u8; 32]; let chunk_hash = match bs58::decode(file_name).onto(&mut hash_buf) { Ok(_) => blake3::Hash::from_bytes(hash_buf), Err(_) => continue, }; // If there is a problem with opening the file, remove it. let Ok(mut chunk_fd) = File::open(&chunk_path).await else { deleted_chunk_paths.insert(chunk_path); deleted_chunks.insert(chunk_hash); continue }; // Perform consistency check let Ok(bytes_read) = read_until_filled(&mut chunk_fd, &mut buf).await else { deleted_chunk_paths.insert(chunk_path); deleted_chunks.insert(chunk_hash); buf = [0u8; MAX_CHUNK_SIZE]; continue }; let chunk_slice = &buf[..bytes_read]; let hashed_chunk = blake3::hash(chunk_slice); // If the hash doesn't match the filename, remove it. if chunk_hash != hashed_chunk { deleted_chunk_paths.insert(chunk_path); deleted_chunks.insert(chunk_hash); buf = [0u8; MAX_CHUNK_SIZE]; continue } // Seems legit. buf = [0u8; MAX_CHUNK_SIZE]; } for chunk_path in &deleted_chunk_paths { if let Err(e) = fs::remove_file(chunk_path).await { warn!( target: "geode::garbage_collect()", "[Geode] Garbage collect failed to remove corrupted chunk: {}", e, ); } } // Perform health check over file metadata. For now we just ensure they // have the correct format. let mut file_paths = fs::read_dir(&self.files_path).await?; while let Some(file) = file_paths.next().await { let Ok(entry) = file else { continue }; let path = entry.path(); // Skip if we're not a plain file if !path.is_file() { continue } // Make sure that the filename is a BLAKE3 hash let file_name = match path.file_name().and_then(|n| n.to_str()) { Some(v) => v, None => continue, }; let mut hash_buf = [0u8; 32]; let file_hash = match bs58::decode(file_name).onto(&mut hash_buf) { Ok(_) => blake3::Hash::from_bytes(hash_buf), Err(_) => continue, }; // The filename is a BLAKE3 hash. It should contain a newline-separated // list of chunks which represent the full file. If that is not the case // we will consider it a corrupted file and delete it. if Self::read_metadata(&path).await.is_err() { if let Err(e) = fs::remove_file(path).await { warn!( target: "geode::garbage_collect()", "[Geode] Garbage collect failed to remove corrupted file: {}", e, ); } deleted_files.insert(file_hash); continue } } info!(target: "geode::garbage_collect()", "[Geode] Garbage collection finished"); Ok((deleted_files, deleted_chunks)) } /// Insert a file into Geode. The function expects any kind of byte stream, which /// can either be another file on the filesystem, a buffer, etc. /// Returns a tuple of `(blake3::Hash, Vec)` which represents the /// file hash, and the file's chunks, respectively. pub async fn insert( &self, mut stream: impl AsyncRead + Unpin, ) -> Result<(blake3::Hash, Vec)> { info!(target: "geode::insert()", "[Geode] Inserting file..."); let mut file_hasher = blake3::Hasher::new(); let mut chunk_hashes = vec![]; let mut buf = [0u8; MAX_CHUNK_SIZE]; loop { let bytes_read = read_until_filled(&mut stream, &mut buf).await?; if bytes_read == 0 { break } let chunk_slice = &buf[..bytes_read]; let chunk_hash = blake3::hash(chunk_slice); file_hasher.update(chunk_hash.as_bytes()); chunk_hashes.push(chunk_hash); // Write the chunk to a file, if necessary. We first perform // a consistency check and if things are fine, we don't have // to perform a write, which is usually more expensive than // reading from disk. let mut chunk_path = self.chunks_path.clone(); chunk_path.push(hash_to_string(&chunk_hash).as_str()); let chunk_fd = OpenOptions::new().read(true).write(true).create(true).open(&chunk_path).await?; let mut fs_buf = [0u8; MAX_CHUNK_SIZE]; let fs_bytes_read = read_until_filled(chunk_fd, &mut fs_buf).await?; let fs_chunk_slice = &fs_buf[..fs_bytes_read]; let fs_chunk_hash = blake3::hash(fs_chunk_slice); if fs_chunk_hash != chunk_hash { debug!( target: "geode::insert()", "Existing chunk inconsistent or unavailable. Writing chunk to {:?}", chunk_path, ); // Here the chunk is broken, so we'll truncate and write the new one. let mut chunk_fd = OpenOptions::new() .read(true) .write(true) .create(true) .open(&chunk_path) .await?; chunk_fd.set_len(0).await?; chunk_fd.seek(SeekFrom::Start(0)).await?; chunk_fd.write_all(chunk_slice).await?; chunk_fd.flush().await?; } else { debug!( target: "geode::insert()", "Existing chunk consistent. Skipping write to {:?}", chunk_path, ); } buf = [0u8; MAX_CHUNK_SIZE]; } // This hash is the file's chunks hashes hashed in order. let file_hash = file_hasher.finalize(); let mut file_path = self.files_path.clone(); file_path.push(hash_to_string(&file_hash).as_str()); // We always overwrite the metadata. let mut file_fd = File::create(&file_path).await?; for ch in &chunk_hashes { file_fd.write(format!("{}\n", hash_to_string(ch).as_str()).as_bytes()).await?; } file_fd.flush().await?; Ok((file_hash, chunk_hashes)) } /// Create and insert file metadata into Geode given a list of hashes. /// Always overwrites any existing file. /// Verifies that the file hash matches the chunk hashes pub async fn insert_file( &self, file_hash: &blake3::Hash, chunk_hashes: &[blake3::Hash], ) -> Result<()> { info!(target: "geode::insert_file()", "[Geode] Inserting file metadata"); if !self.verify_file(file_hash, chunk_hashes) { // The chunk list or file hash is wrong return Err(Error::GeodeNeedsGc) } let mut file_path = self.files_path.clone(); file_path.push(hash_to_string(file_hash).as_str()); let mut file_fd = File::create(&file_path).await?; for ch in chunk_hashes { file_fd.write(format!("{}\n", hash_to_string(ch).as_str()).as_bytes()).await?; } file_fd.flush().await?; Ok(()) } /// Create and insert a single chunk into Geode given a stream. /// Always overwrites any existing chunk. Returns the chunk hash once inserted. pub async fn insert_chunk(&self, stream: impl AsRef<[u8]>) -> Result { info!(target: "geode::insert_chunk()", "[Geode] Inserting single chunk"); let mut cursor = Cursor::new(&stream); let mut chunk = [0u8; MAX_CHUNK_SIZE]; let bytes_read = read_until_filled(&mut cursor, &mut chunk).await?; let chunk_slice = &chunk[..bytes_read]; let chunk_hash = blake3::hash(chunk_slice); let mut chunk_path = self.chunks_path.clone(); chunk_path.push(hash_to_string(&chunk_hash).as_str()); let mut chunk_fd = File::create(&chunk_path).await?; chunk_fd.write_all(chunk_slice).await?; chunk_fd.flush().await?; Ok(chunk_hash) } /// Fetch file metadata from Geode. Returns [`ChunkedFile`] which gives a list /// of chunks and optionally file paths to the said chunks. Returns an error if /// the read failed in any way (could also be the file does not exist). pub async fn get(&self, file_hash: &blake3::Hash) -> Result { let file_hash_str = hash_to_string(file_hash); info!(target: "geode::get()", "[Geode] Getting file chunks for {}...", file_hash_str); let mut file_path = self.files_path.clone(); file_path.push(file_hash_str); // Try to read the file metadata. If it's corrupt, return an error signalling // that garbage collection needs to run. let chunk_hashes = match Self::read_metadata(&file_path).await { Ok(v) => v, Err(e) => { return match e { // If the file is not found, return according error. Error::Io(std::io::ErrorKind::NotFound) => Err(Error::GeodeFileNotFound), // Anything else should tell the client to do garbage collection _ => Err(Error::GeodeNeedsGc), } } }; // Make sure the chunk hashes match with the file hash if !self.verify_file(file_hash, &chunk_hashes) { return Err(Error::GeodeNeedsGc); } let mut chunked_file = ChunkedFile::new(&chunk_hashes); // Iterate over chunks and find which chunks we have available locally. let mut buf = vec![]; for (chunk_hash, chunk_path) in chunked_file.0.iter_mut() { let mut c_path = self.chunks_path.clone(); c_path.push(hash_to_string(chunk_hash).as_str()); if !c_path.exists() || !c_path.is_file() { // TODO: We should be aggressive here and remove the non-file. continue } // Perform chunk consistency check let mut chunk_fd = File::open(&c_path).await?; let bytes_read = chunk_fd.read_to_end(&mut buf).await?; let chunk_slice = &buf[..bytes_read]; let hashed_chunk = blake3::hash(chunk_slice); if &hashed_chunk != chunk_hash { // The chunk is corrupted/inconsistent. Garbage collection should run. buf = vec![]; continue } *chunk_path = Some(c_path); buf = vec![]; } Ok(chunked_file) } /// Fetch a single chunk from Geode. Returns a `PathBuf` pointing to the chunk /// if it is found. pub async fn get_chunk(&self, chunk_hash: &blake3::Hash) -> Result { let chunk_hash_str = hash_to_string(chunk_hash); info!(target: "geode::get_chunk()", "[Geode] Getting chunk {}", chunk_hash_str); let mut chunk_path = self.chunks_path.clone(); chunk_path.push(chunk_hash_str); if !chunk_path.exists() || !chunk_path.is_file() { // TODO: We should be aggressive here and remove the non-file. return Err(Error::GeodeChunkNotFound) } // Perform chunk consistency check let mut buf = vec![]; let mut chunk_fd = File::open(&chunk_path).await?; let bytes_read = chunk_fd.read_to_end(&mut buf).await?; if !self.verify_chunk(chunk_hash, &buf[..bytes_read]) { // The chunk is corrupted return Err(Error::GeodeNeedsGc) } Ok(chunk_path) } /// Verifies that the file hash matches the chunk hashes. pub fn verify_file(&self, file_hash: &blake3::Hash, chunk_hashes: &[blake3::Hash]) -> bool { info!(target: "geode::verify_file()", "[Geode] Verifying file metadata"); let mut file_hasher = blake3::Hasher::new(); for chunk_hash in chunk_hashes { file_hasher.update(chunk_hash.as_bytes()); } *file_hash == file_hasher.finalize() } /// Verifies that the chunk hash matches the content. pub fn verify_chunk(&self, chunk_hash: &blake3::Hash, chunk_slice: &[u8]) -> bool { blake3::hash(chunk_slice) == *chunk_hash } /// Assemble chunks to create a file. /// This method does NOT perform a consistency check. pub async fn assemble_file( &self, file_hash: &blake3::Hash, chunked_file: &ChunkedFile, file_path: &PathBuf, ) -> Result<()> { let file_hash_str = hash_to_string(file_hash); info!(target: "geode::assemble_file()", "[Geode] Assembling file {}", file_hash_str); if file_path.exists() && file_path.is_dir() { return Err(Error::Custom("File path is an existing directory".to_string())) // TODO } let mut file_fd = File::create(&file_path).await?; for (_, chunk_path) in chunked_file.iter() { let mut buf = vec![]; let mut chunk_fd = File::open(chunk_path.clone().unwrap()).await?; let bytes_read = chunk_fd.read_to_end(&mut buf).await?; let chunk_slice = &buf[..bytes_read]; file_fd.write(chunk_slice).await?; file_fd.flush().await?; } Ok(()) } /// List file hashes. pub async fn list_files(&self) -> Result> { info!(target: "geode::list_files()", "[Geode] Listing files"); let mut dir = fs::read_dir(&self.files_path).await?; let mut file_hashes = vec![]; while let Some(file) = dir.try_next().await? { let os_file_name = file.file_name(); let file_name = os_file_name.to_string_lossy().to_string(); let mut hash_buf = [0u8; 32]; let file_hash = match bs58::decode(file_name).onto(&mut hash_buf) { Ok(_) => blake3::Hash::from_bytes(hash_buf), Err(_) => continue, }; file_hashes.push(file_hash); } Ok(file_hashes) } }