tesor - Unnamed repository; edit this file 'description' to name the repository.

diff options


context:
space:
mode:

Diffstat

-rw-r--r--	Cargo.lock	9	+8 −1
-rw-r--r--	Cargo.toml	4	+2 −2
-rw-r--r--	src/controller_zig.rs	7	+5 −2
-rw-r--r--	src/main.rs	8	+6 −2
-rw-r--r--	src/service_storage.rs	280	+192 −88

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

@@ -3315,6 +3315,12 @@ dependencies = [

]

[[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"

source = "registry+https://github.com/rust-lang/crates.io-index"

@@ -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 <me@govorov.online>

// 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<Sqlite> for Id {

+ fn type_info() -> sqlite::SqliteTypeInfo {

+ <Vec<u8> as sqlx::Type<Sqlite>>::type_info()

}

- fn blob_root(&self) -> PathBuf {

- self.root.join("blob")

+ fn compatible(ty: &sqlite::SqliteTypeInfo) -> bool {

+ <Vec<u8> as sqlx::Type<Sqlite>>::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<Self, Box<dyn std::error::Error + Send + Sync>> {

+ let value: Vec<u8> = <Vec<u8> as sqlx::decode::Decode<Sqlite>>::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 <Sqlite as sqlx::database::Database>::ArgumentBuffer<'q>,

+ ) -> Result<IsNull, BoxDynError> {

+ let value = self.0.as_bytes().to_vec();

+ <Vec<u8> as Encode<Sqlite>>::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<Sqlite> for Blob {

fn type_info() -> sqlite::SqliteTypeInfo {

<Vec<u8> as sqlx::Type<Sqlite>>::type_info() // BLOB

@@ -94,7 +101,6 @@ impl sqlx::Type<Sqlite> for Blob {

<Vec<u8> as sqlx::Type<Sqlite>>::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<chrono::Utc>,

- 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<chrono::Utc>,

+ 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<u8> {

+ 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<F, Fut>(&self, mut f: F) -> Result<(), StorageError>

+ where

+ F: FnMut(PathBuf) -> Fut,

+ Fut: std::future::Future<Output = Result<(), StorageError>>,

+ {

+ 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<Sqlite>,

- blobfs: FyleSystem,

}

impl StorageService {

pub async fn new(config: sync::Arc<ConfigService>) -> Result<Self, StorageError> {

- 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> = 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<i64> = 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<u8> = 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<u8> {

- 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()

- }

}