From 667baa71e863a457f3f7278573d200d13b59febd Mon Sep 17 00:00:00 2001 From: Nikolay Govorov Date: Thu, 15 Jan 2026 07:09:56 +0000 Subject: Refactor storage with 2-step writing files --- src/service_storage.rs | 388 +++++++++++++++++++++++++++-------------- 1 file changed, 260 insertions(+), 128 deletions(-) diff --git a/src/service_storage.rs b/src/service_storage.rs index fbf0bca..8a94b79 100644 --- a/src/service_storage.rs +++ b/src/service_storage.rs @@ -1,35 +1,85 @@ // SPDX-FileCopyrightText: 2026 Nikolay Govorov // SPDX-License-Identifier: AGPL-3.0-or-later -use std::path::PathBuf; +// This class stores uploaded files and associated metadata. +// Files are immutable but can be deleted. +// +// A unique file is defined by a `scope` and a `filename`. +// The filename can be any valid UTF-8 string up to 1KB +// in size and does not have to be a valid filesystem name. +// +// The list of files and their metadata are stored in a single +// SQLite database table. Small files are stored in a BLOB column +// in SQLite; for larger files, a checksum is stored in the database, +// and the file itself is stored on disk. +// +// On disk new files are written in two steps: +// - the file is written to a temporary file in the same directory; +// - a transaction is opened and a new file entry is created +// - temp file is renamed to the final filename +// - the transaction is committed. +// +// This scheme guarantees that if a record exists in the table, the file is written. +// However if the server crashes after rename but before commit, an orphan file will remain on the disk. +// To combat this, when the server starts, we check that there is an entry +// in the database for each file on the disk, and we also delete all temporary (non-renamed) files. + +use std::path::{Path, PathBuf}; use std::sync; -use tokio::io::AsyncWriteExt; - +use bytes::Bytes; use sha2::{Digest, Sha256}; use sqlx::{FromRow, Pool, Sqlite, query, query_as, sqlite}; use thiserror::Error; +use tokio::fs; +use tokio::io::AsyncWriteExt; +use tracing::{debug, instrument, warn}; use super::service_config::ConfigService; const SQLITE_POOL_SIZE: u32 = 16; const INLINE_THRESHOLD: usize = 256 * 1024; // 256 KB -fn hash_blob(scope: &str, filename: &str, bytes: &[u8]) -> Vec { - let mut hasher = Sha256::new(); - hasher.update(scope.as_bytes()); - hasher.update(b"/"); - hasher.update(filename.as_bytes()); - hasher.update(b"\0"); - hasher.update(bytes); - hasher.finalize().to_vec() +struct FyleSystem { + root: PathBuf, +} + +impl FyleSystem { + fn new(root: &Path) -> Self { + Self { + root: root.to_path_buf(), + } + } + + fn database(&self) -> PathBuf { + self.root.join("index.sqlite") + } + + fn blob_root(&self) -> PathBuf { + self.root.join("blob") + } + + fn object(&self, scope: &str, file: &str) -> PathBuf { + let mut hasher = Sha256::new(); + hasher.update(scope.as_bytes()); + hasher.update(b"/"); + hasher.update(file.as_bytes()); + hasher.update(b"\0"); + + let hash = hasher.finalize(); + let hash_hex = hex::encode(hash); + self.blob_root() + .join(&hash_hex[0..2]) + .join(&hash_hex[2..4]) + .join(&hash_hex) + } } #[derive(Debug, Clone)] -pub struct Blob(pub bytes::Bytes); +pub struct Blob(pub Bytes); impl std::ops::Deref for Blob { - type Target = bytes::Bytes; + type Target = Bytes; fn deref(&self) -> &Self::Target { &self.0 } @@ -50,7 +100,7 @@ impl<'r> sqlx::decode::Decode<'r, Sqlite> for Blob { value: sqlite::SqliteValueRef<'r>, ) -> Result> { let slice: &'r [u8] = <&'r [u8] as sqlx::decode::Decode>::decode(value)?; - Ok(Blob(bytes::Bytes::copy_from_slice(slice))) + Ok(Blob(Bytes::copy_from_slice(slice))) } } @@ -75,6 +125,9 @@ pub enum StorageError { #[error("blob integrity check failed")] IntegrityError, + #[error("blob file missing on disk: {0}")] + BlobNotFound(PathBuf), + #[error("failed to connect index db: {0}")] DbError(#[from] sqlx::Error), @@ -84,44 +137,75 @@ pub enum StorageError { pub struct StorageService { sqlite: Pool, - storage_dir: PathBuf, + blobfs: FyleSystem, } impl StorageService { pub async fn new(config: sync::Arc) -> Result { - let storage_dir = config.dirname().to_path_buf(); - let connection = format!( - "sqlite:{}?mode=rwc", - storage_dir.join("index.sqlite").to_str().unwrap() - ); + let blobfs = FyleSystem::new(config.dirname()); - let pool: Pool = sqlite::SqlitePoolOptions::new() + let connection = format!("sqlite:{}?mode=rwc", blobfs.database().to_str().unwrap()); + let sqlite: Pool = sqlite::SqlitePoolOptions::new() .max_connections(SQLITE_POOL_SIZE) + .after_connect(|conn, _meta| { + Box::pin(async move { + // Connection-specific PRAGMAs (must be set on each connection) + sqlx::query("PRAGMA foreign_keys = ON;") + .execute(&mut *conn) + .await?; + sqlx::query("PRAGMA busy_timeout = 5000;") + .execute(&mut *conn) + .await?; + Ok(()) + }) + }) .connect(&connection) .await?; - let storage = Self { - sqlite: pool, - storage_dir, - }; - storage.msetupdb().await?; + // WAL mode is database-wide and persists, only needs to be set once + query("PRAGMA journal_mode = WAL;").execute(&sqlite).await?; + + let storage = Self { sqlite, blobfs }; + storage.migrations().await?; + storage.doctor().await?; Ok(storage) } - async fn msetupdb(&self) -> Result<(), StorageError> { - query("PRAGMA foreign_keys = ON;") - .execute(&self.sqlite) - .await?; - query("PRAGMA journal_mode = WAL;") - .execute(&self.sqlite) - .await?; - query("PRAGMA busy_timeout = 5000;") - .execute(&self.sqlite) - .await?; + /// Synchronously traverses the tree and removes temporary files. + /// Must run before the application starts. + async fn doctor(&self) -> Result<(), StorageError> { + let blob_root = self.blobfs.blob_root(); + if !blob_root.exists() { + return Ok(()); + } + + let mut stack = vec![blob_root.clone()]; + while let Some(dir) = stack.pop() { + let mut entries = fs::read_dir(&dir).await?; + while let Some(entry) = entries.next_entry().await? { + let path = entry.path(); + let file_type = entry.file_type().await?; + + if file_type.is_dir() { + stack.push(path); + } else if file_type.is_file() { + let name = entry.file_name().to_string_lossy().to_string(); + if name.ends_with(".part") { + warn!(path = %path.display(), "cleanup: removing temp file"); + let _ = fs::remove_file(&path).await; + } + } + } + } + + Ok(()) + } + async fn migrations(&self) -> Result<(), StorageError> { query( - "CREATE TABLE IF NOT EXISTS datafiles( + " + CREATE TABLE IF NOT EXISTS datafiles( id INTEGER PRIMARY KEY, scope TEXT NOT NULL, created_at TEXT DEFAULT (datetime('now')), @@ -131,7 +215,8 @@ impl StorageService { inlined INTEGER NOT NULL, UNIQUE (scope, file_name) - ) STRICT;", + ) STRICT; + ", ) .execute(&self.sqlite) .await?; @@ -139,123 +224,170 @@ impl StorageService { Ok(()) } + #[instrument(skip(self))] + pub async fn get(&self, scope: &str, filename: &str) -> Result, StorageError> { + let file: Option = + query_as("SELECT * FROM datafiles WHERE scope = ?1 AND file_name = ?2") + .bind(scope) + .bind(filename) + .fetch_optional(&self.sqlite) + .await?; + + match file { + None => { + debug!("get: file not found"); + Ok(None) + } + Some(mut file) if !file.inlined => { + let obj = self.blobfs.object(scope, filename); + let bytes = match fs::read(&obj).await { + Ok(b) => Bytes::from(b), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + return Err(StorageError::BlobNotFound(obj)); + } + Err(e) => return Err(e.into()), + }; + + let hash = self.blob_hash(scope, filename, &bytes); + if hash != file.file_bytes.to_vec() { + return Err(StorageError::IntegrityError); + } + + file.file_bytes = Blob(bytes); + debug!(size = file.file_size, "get: loaded from disk"); + Ok(Some(file)) + } + Some(file) => { + debug!(size = file.file_size, "get: loaded inline"); + Ok(Some(file)) + } + } + } + + #[instrument(skip(self, bytes), fields(size = bytes.len()))] pub async fn put( &self, scope: &str, filename: &str, - bytes: &bytes::Bytes, + bytes: &Bytes, ) -> Result<(), StorageError> { let inlined = bytes.len() <= INLINE_THRESHOLD; - let stored_bytes: Vec = if inlined { + let obj = self.blobfs.object(scope, filename); + let tmp = obj.with_extension(format!("{}.part", uuid::Uuid::new_v4())); + + // write temp file + if !inlined { + if let Some(parent) = obj.parent() { + fs::create_dir_all(parent).await?; + } + + let mut file = fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&tmp) + .await?; + + file.write_all(bytes).await?; + file.sync_all().await?; + } + + // bytes for small files or hash for large ones + let payload: Vec = if inlined { bytes.to_vec() } else { - self.write_blob(scope, filename, bytes).await? + self.blob_hash(scope, filename, bytes) }; - let result = query("INSERT INTO datafiles (scope, file_name, file_size, file_bytes, inlined) VALUES (?1, ?2, ?3, ?4, ?5);") + let result = async { + let mut tx = self.sqlite.begin().await?; + + let result = query( + " + INSERT INTO datafiles ( + scope, file_name, file_size, file_bytes, inlined + ) VALUES ( + ?1, ?2, ?3, ?4, ?5 + ); + ", + ) .bind(scope) .bind(filename) .bind(bytes.len() as i64) - .bind(&stored_bytes) + .bind(&payload) .bind(inlined) - .execute(&self.sqlite) + .execute(tx.as_mut()) .await; - match result { - Ok(_) => Ok(()), - Err(e) => { - if !inlined { - let _ = self.delete_blob(&stored_bytes).await; + match result { + Ok(_) => { + if !inlined { + fs::rename(&tmp, &obj).await?; + if let Some(parent) = obj.parent() { + let dir = fs::File::open(parent).await?; + dir.sync_all().await?; + } + } + + tx.commit().await?; + debug!("put: a new file has been commited"); + Ok(()) } - match e { - sqlx::Error::Database(ref db_err) if db_err.is_unique_violation() => Err( - StorageError::AlreadyExists(scope.to_string(), filename.to_string()), - ), - _ => Err(e.into()), + Err(sqlx::Error::Database(ref db_err)) if db_err.is_unique_violation() => { + let existing: Option = + query_as("SELECT * FROM datafiles WHERE scope = ?1 AND file_name = ?2") + .bind(scope) + .bind(filename) + .fetch_optional(tx.as_mut()) + .await?; + + match existing { + Some(file) if file.file_bytes.to_vec() == payload => { + if !inlined { + let _ = fs::remove_file(&tmp).await; + } + + debug!("put: identical file already exists"); + return Ok(()); + } + Some(_) => { + warn!("put: file already exists with different content"); + } + None => { + // unique_violation but row doesn't exist - shouldn't happen + warn!("corrupted! 'is_unique_violation' received, but data cannot be selected"); + } + } + + Err(StorageError::AlreadyExists( + scope.to_string(), + filename.to_string(), + )) } + Err(e) => Err(e.into()), } - } - } + }.await; - async fn write_blob( - &self, - scope: &str, - filename: &str, - bytes: &bytes::Bytes, - ) -> Result, StorageError> { - let hash = hash_blob(scope, filename, bytes); - let hash_hex = hex::encode(&hash); - - let blob_dir = self.storage_dir.join(&hash_hex[0..2]).join(&hash_hex[2..4]); - let blob_path = blob_dir.join(&hash_hex); - - tokio::fs::create_dir_all(&blob_dir).await?; - - let mut file = tokio::fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(&blob_path) - .await - .map_err(|e| { - if e.kind() == std::io::ErrorKind::AlreadyExists { - StorageError::AlreadyExists(scope.to_string(), filename.to_string()) - } else { - StorageError::IoError(e) - } - })?; - - file.write_all(bytes).await?; - file.sync_all().await?; - - Ok(hash) - } - - fn blob_path(&self, hash: &[u8]) -> PathBuf { - let hash_hex = hex::encode(hash); - self.storage_dir - .join(&hash_hex[0..2]) - .join(&hash_hex[2..4]) - .join(&hash_hex) - } + if result.is_err() && !inlined { + let _ = fs::remove_file(&tmp).await; + } - async fn delete_blob(&self, hash: &[u8]) -> Result<(), StorageError> { - tokio::fs::remove_file(self.blob_path(hash)).await?; - Ok(()) + result } - pub async fn get(&self, scope: &str, filename: &str) -> Result, StorageError> { - let file: Option = - query_as("SELECT * FROM datafiles WHERE scope = ?1 AND file_name = ?2") - .bind(scope) - .bind(filename) - .fetch_optional(&self.sqlite) - .await?; + fn blob_hash(&self, scope: &str, file: &str, blob: &[u8]) -> Vec { + let mut hasher = Sha256::new(); - match file { - None => Ok(None), - Some(file) if file.inlined => Ok(Some(file)), - Some(mut file) => { - let bytes = self.read_blob(scope, filename, &file.file_bytes).await?; - file.file_bytes = Blob(bytes); - Ok(Some(file)) - } - } - } + hasher.update(b"/"); + hasher.update(scope.as_bytes()); - async fn read_blob( - &self, - scope: &str, - filename: &str, - expected_hash: &[u8], - ) -> Result { - let bytes = tokio::fs::read(self.blob_path(expected_hash)).await?; + hasher.update(b"/"); + hasher.update(file.as_bytes()); - let actual_hash = hash_blob(scope, filename, &bytes); - if actual_hash != expected_hash { - return Err(StorageError::IntegrityError); - } + hasher.update(b"/"); + hasher.update(blob); + hasher.update(b"\0"); - Ok(bytes::Bytes::from(bytes)) + hasher.finalize().to_vec() } } -- Gilti