From 8001852bf353f38d2428f4a28aab32570ae73754 Mon Sep 17 00:00:00 2001 From: Nikolay Govorov Date: Thu, 15 Jan 2026 09:23:35 +0000 Subject: Use uuid v5 instead of hash for object id --- Cargo.lock | 9 +- Cargo.toml | 4 +- src/controller_zig.rs | 7 +- src/main.rs | 8 +- src/service_storage.rs | 280 ++++++++++++++++++++++++++++------------- 5 files changed, 213 insertions(+), 95 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 056f67e..bd1d6bc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3314,6 +3314,12 @@ dependencies = [ "sha1", ] +[[package]] +name = "sha1_smol" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d" + [[package]] name = "sha2" version = "0.10.9" @@ -4261,6 +4267,7 @@ checksum = "e2e054861b4bd027cd373e18e8d8d8e6548085000e41290d95ce0c373a654b4a" dependencies = [ "getrandom 0.3.4", "js-sys", + "sha1_smol", "wasm-bindgen", ] @@ -4940,6 +4947,7 @@ dependencies = [ "bytes", "cargo-deny", "chrono", + "crc32fast", "hex", "http-body-util", "hyper", @@ -4948,7 +4956,6 @@ dependencies = [ "minijinja", "semver", "serde", - "sha2", "sqlx", "thiserror 2.0.17", "tokio", diff --git a/Cargo.toml b/Cargo.toml index 17953b2..11e5f9a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,7 @@ repository = "https://github.com/mrdimidium/Zorian" axum = { version = "0.8.8", features = ["http2", "macros", "multipart"] } bytes = "1.11.0" chrono = "0.4" +crc32fast = "1.4.2" hex = "0.4" http-body-util = "0.1.3" hyper = "1.8.1" @@ -20,7 +21,6 @@ hyper-util = { version = "0.1.19", features = ["client", "http1", "http2", "toki minijinja = "2.14.0" semver = "1.0" serde = { version = "1.0", features = ["derive"] } -sha2 = "0.10" sqlx = { version = "0.8.6", features = ["sqlite", "runtime-tokio", "macros", "derive", "chrono", "json"] } thiserror = "2.0" tokio = { version = "1.49.0", features = ["full"] } @@ -29,7 +29,7 @@ tower-http = { version = "0.6.8", features = ["full"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } tower = "0.5.3" -uuid = { version = "1.19.0", features = ["v4"] } +uuid = { version = "1.19.0", features = ["v4","v5"] } [dev-dependencies] cargo-deny = "0.19.0" diff --git a/src/controller_zig.rs b/src/controller_zig.rs index ba33462..7e9e6a8 100644 --- a/src/controller_zig.rs +++ b/src/controller_zig.rs @@ -1,10 +1,12 @@ // SPDX-FileCopyrightText: 2026 Nikolay Govorov // SPDX-License-Identifier: AGPL-3.0-or-later +use std::sync::Arc; + use axum::{Router, body, extract, http, response, routing}; use semver::Version; -use std::sync::Arc; use thiserror::Error; +use tracing::error; use crate::service_config; use crate::service_storage; @@ -179,7 +181,8 @@ impl ZigController { )); } Ok(None) => {} - Err(_) => { + Err(err) => { + error!("failed get file from storage: {err}"); return Err(http::StatusCode::INTERNAL_SERVER_ERROR); } } diff --git a/src/main.rs b/src/main.rs index 2c68dfb..6d74b09 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,7 +5,10 @@ use std::path::PathBuf; use std::sync::Arc; use axum::{Router, http}; -use tower_http::{request_id, trace::TraceLayer}; +use tower_http::{ + request_id, + trace::{DefaultOnFailure, TraceLayer}, +}; use tracing::{error, info}; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; @@ -103,7 +106,8 @@ async fn main() { tracing::info_span!("http_request", request_id = %request_id) }) .on_request(log_request) - .on_response(log_response); + .on_response(log_response) + .on_failure(DefaultOnFailure::new().level(tracing::Level::ERROR)); let app = Router::new() .merge(web_controller.router()) diff --git a/src/service_storage.rs b/src/service_storage.rs index 8a94b79..f1f04b2 100644 --- a/src/service_storage.rs +++ b/src/service_storage.rs @@ -28,63 +28,70 @@ use std::path::{Path, PathBuf}; use std::sync; use bytes::Bytes; -use sha2::{Digest, Sha256}; -use sqlx::{FromRow, Pool, Sqlite, query, query_as, sqlite}; +use crc32fast::Hasher as Crc32Hasher; +use sqlx::encode::{Encode, IsNull}; +use sqlx::error::BoxDynError; +use sqlx::{FromRow, Pool, Sqlite, query, query_as, query_scalar, sqlite}; use thiserror::Error; use tokio::fs; use tokio::io::AsyncWriteExt; use tracing::{debug, instrument, warn}; +use uuid::Uuid; use super::service_config::ConfigService; const SQLITE_POOL_SIZE: u32 = 16; const INLINE_THRESHOLD: usize = 256 * 1024; // 256 KB -struct FyleSystem { - root: PathBuf, -} +// You cannot change this ID, this will desynchronize the records in the database and the files on the disk. +const UUID_ROOT_NAMESPACE: Uuid = Uuid::from_bytes([ + 0x8b, 0x06, 0x3c, 0x4c, 0x6b, 0x5c, 0x4a, 0x8b, 0x92, 0x8f, 0x75, 0x8b, 0x0e, 0x63, 0xc3, 0x5d, +]); -impl FyleSystem { - fn new(root: &Path) -> Self { - Self { - root: root.to_path_buf(), - } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct Id(pub Uuid); +impl std::ops::Deref for Id { + type Target = Uuid; + fn deref(&self) -> &Self::Target { + &self.0 } - - fn database(&self) -> PathBuf { - self.root.join("index.sqlite") +} +impl sqlx::Type for Id { + fn type_info() -> sqlite::SqliteTypeInfo { + as sqlx::Type>::type_info() } - fn blob_root(&self) -> PathBuf { - self.root.join("blob") + fn compatible(ty: &sqlite::SqliteTypeInfo) -> bool { + as sqlx::Type>::compatible(ty) } - - 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) +} +impl<'r> sqlx::decode::Decode<'r, Sqlite> for Id { + fn decode( + value: sqlite::SqliteValueRef<'r>, + ) -> Result> { + let value: Vec = as sqlx::decode::Decode>::decode(value)?; + let uuid = Uuid::from_slice(&value)?; + Ok(Id(uuid)) + } +} +impl<'q> Encode<'q, Sqlite> for Id { + fn encode_by_ref( + &self, + buf: &mut ::ArgumentBuffer<'q>, + ) -> Result { + let value = self.0.as_bytes().to_vec(); + as Encode>::encode(value, buf) } } #[derive(Debug, Clone)] pub struct Blob(pub Bytes); - impl std::ops::Deref for Blob { type Target = Bytes; fn deref(&self) -> &Self::Target { &self.0 } } - impl sqlx::Type for Blob { fn type_info() -> sqlite::SqliteTypeInfo { as sqlx::Type>::type_info() // BLOB @@ -94,7 +101,6 @@ impl sqlx::Type for Blob { as sqlx::Type>::compatible(ty) } } - impl<'r> sqlx::decode::Decode<'r, Sqlite> for Blob { fn decode( value: sqlite::SqliteValueRef<'r>, @@ -104,19 +110,6 @@ impl<'r> sqlx::decode::Decode<'r, Sqlite> for Blob { } } -#[allow(unused)] -#[derive(Debug, Clone, FromRow)] -pub struct File { - pub id: u64, - pub scope: String, - pub created_at: chrono::DateTime, - - pub file_name: String, - pub file_size: i64, - pub file_bytes: Blob, - pub inlined: bool, -} - #[derive(Debug, Error)] pub enum StorageError { #[error("file already exists: {0}/{1}")] @@ -135,14 +128,108 @@ pub enum StorageError { IoError(#[from] std::io::Error), } +#[allow(unused)] +#[derive(Debug, Clone, FromRow)] +pub struct File { + pub id: Id, + pub scope: String, + pub created_at: chrono::DateTime, + + pub file_name: String, + pub file_size: i64, + pub file_bytes: Blob, + pub inlined: bool, +} + +impl File { + fn uuid(scope: &str, file: &str) -> Id { + let scope_ns = Uuid::new_v5(&UUID_ROOT_NAMESPACE, scope.as_bytes()); + Id(Uuid::new_v5(&scope_ns, file.as_bytes())) + } + + fn hash(scope: &str, file: &str, blob: &[u8]) -> Vec { + let mut hasher = Crc32Hasher::new(); + + hasher.update(b"/"); + hasher.update(scope.as_bytes()); + + hasher.update(b"/"); + hasher.update(file.as_bytes()); + + hasher.update(b"/"); + hasher.update(blob); + hasher.update(b"\0"); + + hasher.finalize().to_be_bytes().to_vec() + } +} + +struct FileSystem { + objects: PathBuf, + database: PathBuf, +} + +impl FileSystem { + fn new(root: &Path) -> Self { + Self { + objects: root.join("objects"), + database: root.join("index.sqlite"), + } + } + + fn database(&self) -> &Path { + &self.database + } + + fn objects_root(&self) -> &Path { + &self.objects + } + + fn object(&self, scope: &str, file: &str) -> PathBuf { + let id = File::uuid(scope, file); + let hash_hex = hex::encode(id.0.as_bytes()); + self.objects_root() + .join(&hash_hex[0..2]) + .join(&hash_hex[2..4]) + .join(&hash_hex) + } + + async fn objects_walk(&self, mut f: F) -> Result<(), StorageError> + where + F: FnMut(PathBuf) -> Fut, + Fut: std::future::Future>, + { + if !self.objects.exists() { + return Ok(()); + } + + let mut stack = vec![self.objects.to_path_buf()]; + 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() { + f(path).await?; + } + } + } + + Ok(()) + } +} + pub struct StorageService { + blobfs: FileSystem, sqlite: Pool, - blobfs: FyleSystem, } impl StorageService { pub async fn new(config: sync::Arc) -> Result { - let blobfs = FyleSystem::new(config.dirname()); + let blobfs = FileSystem::new(config.dirname()); let connection = format!("sqlite:{}?mode=rwc", blobfs.database().to_str().unwrap()); let sqlite: Pool = sqlite::SqlitePoolOptions::new() @@ -175,29 +262,59 @@ impl StorageService { /// 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 result = self.blobfs.objects_walk(|path| async move { + let name = match path.file_name().and_then(|name| name.to_str()) { + Some(name) => name.to_string(), + None => { + warn!(path = %path.display(), "cleanup: invalid blob name"); + let _ = fs::remove_file(path).await; + 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 name.ends_with(".part") { + warn!(path = %path.display(), "cleanup: removing temp file"); + let _ = fs::remove_file(path).await; + return Ok(()); + } - 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; - } + if name.len() != 32 { + warn!(path = %path.display(), "cleanup: invalid blob name length"); + let _ = fs::remove_file(path).await; + return Ok(()); + } + + let id_bytes = match hex::decode(&name) { + Ok(bytes) => bytes, + Err(_) => { + warn!(path = %path.display(), "cleanup: invalid blob name hex"); + let _ = fs::remove_file(path).await; + return Ok(()); + } + }; + + let id = match Uuid::from_slice(&id_bytes) { + Ok(uuid) => Id(uuid), + Err(_) => { + warn!(path = %path.display(), "cleanup: invalid blob uuid"); + let _ = fs::remove_file(path).await; + return Ok(()); } + }; + + let exists: Option = query_scalar("SELECT 1 FROM datafiles WHERE id = ?1") + .bind(id) + .fetch_optional(&self.sqlite) + .await?; + + if exists.is_none() { + warn!(path = %path.display(), "cleanup: removing orphan blob"); + let _ = fs::remove_file(path).await; } - } + + Ok(()) + }); + result.await?; Ok(()) } @@ -206,7 +323,7 @@ impl StorageService { query( " CREATE TABLE IF NOT EXISTS datafiles( - id INTEGER PRIMARY KEY, + id BLOB PRIMARY KEY CHECK (length(id) = 16), scope TEXT NOT NULL, created_at TEXT DEFAULT (datetime('now')), file_name TEXT NOT NULL, @@ -248,7 +365,7 @@ impl StorageService { Err(e) => return Err(e.into()), }; - let hash = self.blob_hash(scope, filename, &bytes); + let hash = File::hash(scope, filename, &bytes); if hash != file.file_bytes.to_vec() { return Err(StorageError::IntegrityError); } @@ -274,7 +391,7 @@ impl StorageService { let inlined = bytes.len() <= INLINE_THRESHOLD; let obj = self.blobfs.object(scope, filename); - let tmp = obj.with_extension(format!("{}.part", uuid::Uuid::new_v4())); + let tmp = obj.with_extension(format!("{}.part", Uuid::new_v4())); // write temp file if !inlined { @@ -296,21 +413,23 @@ impl StorageService { let payload: Vec = if inlined { bytes.to_vec() } else { - self.blob_hash(scope, filename, bytes) + File::hash(scope, filename, bytes) }; let result = async { let mut tx = self.sqlite.begin().await?; + let id = File::uuid(scope, filename); let result = query( " INSERT INTO datafiles ( - scope, file_name, file_size, file_bytes, inlined + id, scope, file_name, file_size, file_bytes, inlined ) VALUES ( - ?1, ?2, ?3, ?4, ?5 + ?1, ?2, ?3, ?4, ?5, ?6 ); ", ) + .bind(id) .bind(scope) .bind(filename) .bind(bytes.len() as i64) @@ -345,6 +464,7 @@ impl StorageService { Some(file) if file.file_bytes.to_vec() == payload => { if !inlined { let _ = fs::remove_file(&tmp).await; + tx.rollback().await?; } debug!("put: identical file already exists"); @@ -374,20 +494,4 @@ impl StorageService { result } - - fn blob_hash(&self, scope: &str, file: &str, blob: &[u8]) -> Vec { - let mut hasher = Sha256::new(); - - hasher.update(b"/"); - hasher.update(scope.as_bytes()); - - hasher.update(b"/"); - hasher.update(file.as_bytes()); - - hasher.update(b"/"); - hasher.update(blob); - hasher.update(b"\0"); - - hasher.finalize().to_vec() - } } -- Gilti