cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

perf(round26): drive-policy JSONB decode, CachedBlobBackend shard-dir pre-create, delta-upload foldhash

Browse Source 
Three benchmark-gated optimizations from the ROUND25 backlog (benches/ROUND26.md),
each with a BEFORE/AFTER gate that rolls back if AFTER does not beat BEFORE:

- P1 drive_pg_repository policy reads: decode d.policies through
  sqlx::types::Json<DrivePolicies> (one from_slice over the raw JSONB bytes)
  instead of a throwaway serde_json::Value DOM + DrivePolicies::from_value —
  6 -> 0 allocs/read, 2.77x wall. A shared policies_from_row helper preserves
  the lenient unwrap_or_default fallback (malformed bag -> all-false).
- D1 CachedBlobBackend: pre-create the 256 {00..ff} shard dirs at initialize()
  (mirroring LocalBlobBackend, reusing HEX_PREFIXES) and drop the redundant
  per-write create_dir_all on already-existing shards — ~45us + a blocking-pool
  dispatch removed per cache write on cached-remote deployments.
- G1 delta-upload have/need hash sets (distinct_hashes, authorize_chunk_download):
  SipHash -> foldhash::quality::RandomState — a fast hasher that stays
  DoS-resistant via a per-instance random seed, the required property for the
  attacker-controlled 64-hex client hashes — 2.37x wall on a 40k-hash
  negotiation. foldhash was already in the lockfile transitively (hashbrown).

Tested and REVERTED (kept as-is): moving the moka eviction unlink off the reactor
via spawn_blocking. The benchmark refuted it — on the local cache dir the
spawn_blocking dispatch (~20us) costs more than the inline unlink (~7us) it would
replace. See ROUND26.md §D2.

Adds bench_round26_{micro,diskio,hasher} (counting allocator / async wall / wall).
Verified: cargo fmt clean, cargo clippy --features bench -D warnings clean,
cargo test --lib --features bench = 529 passed / 0 failed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01L8gs91AhmazoxMsDcNk3KT
This commit is contained in:
Claude
2026-07-21 01:05:12 +00:00
parent e8e4ef4b15
commit 5b2bb8f883
10 changed files with 631 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/application/services/delta_upload_service.rs
+7 -2
View File
@@ -553,7 +553,10 @@ impl DeltaUploadService {
self.max_chunk_count()
)));
}
let mut distinct_seen = HashSet::new();
// foldhash::quality::RandomState — a fast, per-instance random-seeded
// hasher, DoS-safe for these attacker-controlled client hashes (up to
// max_chunk_count() of them per request) — benches/ROUND26.md §G1.
let mut distinct_seen: HashSet<&str, foldhash::quality::RandomState> = HashSet::default();
for hash in &request.hashes {
if !is_valid_hash(hash) {
return Err(DomainError::validation_error(
@@ -684,7 +687,9 @@ fn sanitize_file_name(name: &str) -> Result<String, DomainError> {
/// Distinct hashes in first-occurrence order.
fn distinct_hashes(chunks: &[ChunkRef]) -> Vec<String> {
let mut seen = HashSet::new();
// foldhash::quality::RandomState — fast, per-instance random-seeded and thus
// DoS-safe for these attacker-controlled client hashes (benches/ROUND26.md §G1).
let mut seen: HashSet<&str, foldhash::quality::RandomState> = HashSet::default();
chunks
.iter()
.filter(|c| seen.insert(c.h.as_str()))
src/infrastructure/repositories/pg/drive_pg_repository.rs
+35 -32
View File
@@ -20,6 +20,19 @@ use crate::domain::repositories::drive_repository::{
DriveRepository, DriveRepositoryError, DriveWithRootName,
};
/// Decode a `d.policies` JSONB column straight into `DrivePolicies` via
/// `sqlx::types::Json<T>` — a single `serde_json::from_slice` over the raw JSONB
/// bytes — instead of fetching a throwaway `serde_json::Value` DOM and walking it
/// once with `DrivePolicies::from_value`. The §J1 pattern (ROUND23) applied to
/// the drive-policy path §J2 left behind (benches/ROUND26.md §P1). The lenient
/// `unwrap_or_default` fallback (a malformed bag decodes to all-false rather than
/// erroring the read) is preserved exactly.
fn policies_from_row(row: &sqlx::postgres::PgRow) -> crate::domain::entities::drive::DrivePolicies {
row.try_get::<sqlx::types::Json<crate::domain::entities::drive::DrivePolicies>, _>("policies")
.map(|j| j.0)
.unwrap_or_default()
}
/// `default_drive_cache` TTL. The default-drive → root-folder binding is
/// nearly immutable (changes only on provisioning / drive deletion /
/// policy edits — all of which invalidate explicitly below), yet it is
@@ -706,7 +719,7 @@ impl DriveRepository for DrivePgRepository {
&self,
file_id: Uuid,
) -> Result<crate::domain::entities::drive::DrivePolicies, DriveRepositoryError> {
let row: Option<(serde_json::Value,)> = sqlx::query_as(
let row = sqlx::query(
"SELECT d.policies \
FROM storage.drives d \
JOIN storage.files f ON f.drive_id = d.id \
@@ -715,20 +728,16 @@ impl DriveRepository for DrivePgRepository {
.bind(file_id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| Self::map_sqlx_err("get_policies_for_file", e))?;
let raw = row
.ok_or_else(|| DriveRepositoryError::NotFound(file_id.to_string()))?
.0;
Ok(crate::domain::entities::drive::DrivePolicies::from_value(
&raw,
))
.map_err(|e| Self::map_sqlx_err("get_policies_for_file", e))?
.ok_or_else(|| DriveRepositoryError::NotFound(file_id.to_string()))?;
Ok(policies_from_row(&row))
}
async fn get_policies_for_folder(
&self,
folder_id: Uuid,
) -> Result<crate::domain::entities::drive::DrivePolicies, DriveRepositoryError> {
let row: Option<(serde_json::Value,)> = sqlx::query_as(
let row = sqlx::query(
"SELECT d.policies \
FROM storage.drives d \
JOIN storage.folders fo ON fo.drive_id = d.id \
@@ -737,20 +746,16 @@ impl DriveRepository for DrivePgRepository {
.bind(folder_id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| Self::map_sqlx_err("get_policies_for_folder", e))?;
let raw = row
.ok_or_else(|| DriveRepositoryError::NotFound(folder_id.to_string()))?
.0;
Ok(crate::domain::entities::drive::DrivePolicies::from_value(
&raw,
))
.map_err(|e| Self::map_sqlx_err("get_policies_for_folder", e))?
.ok_or_else(|| DriveRepositoryError::NotFound(folder_id.to_string()))?;
Ok(policies_from_row(&row))
}
async fn get_drive_id_and_policies_for_file(
&self,
file_id: Uuid,
) -> Result<(Uuid, crate::domain::entities::drive::DrivePolicies), DriveRepositoryError> {
let row: Option<(Uuid, serde_json::Value)> = sqlx::query_as(
let row = sqlx::query(
"SELECT d.id, d.policies \
FROM storage.drives d \
JOIN storage.files f ON f.drive_id = d.id \
@@ -759,20 +764,19 @@ impl DriveRepository for DrivePgRepository {
.bind(file_id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_file", e))?;
let (drive_id, raw) =
row.ok_or_else(|| DriveRepositoryError::NotFound(file_id.to_string()))?;
Ok((
drive_id,
crate::domain::entities::drive::DrivePolicies::from_value(&raw),
))
.map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_file", e))?
.ok_or_else(|| DriveRepositoryError::NotFound(file_id.to_string()))?;
let drive_id: Uuid = row
.try_get("id")
.map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_file", e))?;
Ok((drive_id, policies_from_row(&row)))
}
async fn get_drive_id_and_policies_for_folder(
&self,
folder_id: Uuid,
) -> Result<(Uuid, crate::domain::entities::drive::DrivePolicies), DriveRepositoryError> {
let row: Option<(Uuid, serde_json::Value)> = sqlx::query_as(
let row = sqlx::query(
"SELECT d.id, d.policies \
FROM storage.drives d \
JOIN storage.folders fo ON fo.drive_id = d.id \
@@ -781,13 +785,12 @@ impl DriveRepository for DrivePgRepository {
.bind(folder_id)
.fetch_optional(self.pool.as_ref())
.await
.map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_folder", e))?;
let (drive_id, raw) =
row.ok_or_else(|| DriveRepositoryError::NotFound(folder_id.to_string()))?;
Ok((
drive_id,
crate::domain::entities::drive::DrivePolicies::from_value(&raw),
))
.map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_folder", e))?
.ok_or_else(|| DriveRepositoryError::NotFound(folder_id.to_string()))?;
let drive_id: Uuid = row
.try_get("id")
.map_err(|e| Self::map_sqlx_err("get_drive_id_and_policies_for_folder", e))?;
Ok((drive_id, policies_from_row(&row)))
}
async fn drive_id_for_folder(&self, folder_id: Uuid) -> Result<Uuid, DriveRepositoryError> {
src/infrastructure/services/cached_blob_backend.rs
+16 -14
View File
@@ -119,10 +119,21 @@ impl BlobStorageBackend for CachedBlobBackend {
Box::pin(async move {
inner.initialize().await?;
// Create cache dir structure (256 prefix dirs)
// Create the cache dir AND its 256 {00..ff} shard dirs up front
// (mirroring LocalBlobBackend::initialize), so the write paths never
// pay a per-chunk `create_dir_all` on an already-existing shard — a
// ~45 µs mkdirat(EEXIST)+stat+blocking-dispatch removed per cache
// write on cached-remote deployments (benches/ROUND26.md §D1).
fs::create_dir_all(&cache_dir).await.map_err(|e| {
DomainError::internal_error("BlobCache", format!("mkdir cache_dir: {e}"))
})?;
for prefix in &crate::infrastructure::services::local_blob_backend::HEX_PREFIXES {
fs::create_dir_all(cache_dir.join(prefix))
.await
.map_err(|e| {
DomainError::internal_error("BlobCache", format!("mkdir cache shard: {e}"))
})?;
}
// Scan existing cache to rebuild index. Collect entries WITHOUT
// holding the index lock — a large cache directory walk must not
@@ -411,10 +422,9 @@ impl CachedBlobBackend {
/// index deliberately skipped the eviction sweep on this path, letting
/// write bursts overshoot the budget until the next read-miss insert).
async fn cache_bytes_write_through(&self, hash: String, data: &Bytes) {
// The shard dir was created at initialize() — no per-write create_dir_all
// (benches/ROUND26.md §D1).
let dest = self.cached_path(&hash);
if let Some(parent) = dest.parent() {
let _ = fs::create_dir_all(parent).await;
}
let _ = fs::write(&dest, data).await;
let data_len = data.len() as u64;
self.index.insert(hash, CacheEntry { size: data_len });
@@ -451,12 +461,8 @@ impl CachedBlobBackend {
}
async fn insert_into_cache(&self, hash: &str, source_path: &Path) -> Result<(), DomainError> {
// Shard dir pre-created at initialize() (benches/ROUND26.md §D1).
let dest = self.cached_path(hash);
if let Some(parent) = dest.parent() {
fs::create_dir_all(parent).await.map_err(|e| {
DomainError::internal_error("BlobCache", format!("mkdir failed: {e}"))
})?;
}
let size = fs::metadata(source_path)
.await
@@ -476,12 +482,8 @@ impl CachedBlobBackend {
async fn fetch_and_cache(&self, hash: &str) -> Result<PathBuf, DomainError> {
let stream = self.inner.get_blob_stream(hash).await?;
// Shard dir pre-created at initialize() (benches/ROUND26.md §D1).
let dest = self.cached_path(hash);
if let Some(parent) = dest.parent() {
fs::create_dir_all(parent).await.map_err(|e| {
DomainError::internal_error("BlobCache", format!("mkdir failed: {e}"))
})?;
}
// Unique temp name: even if two fetches for one hash ever race
// (e.g. across processes sharing a cache dir), each writes its own
src/infrastructure/services/local_blob_backend.rs
+1 -1
View File
@@ -164,7 +164,7 @@ pub async fn write_blob_bytes_for_bench(
}
/// Compile-time lookup table for the 256 two-digit lowercase hex prefixes ("00"…"ff").
static HEX_PREFIXES: [&str; 256] = [
pub(crate) static HEX_PREFIXES: [&str; 256] = [
"00", "01", "02", "03", "04", "05", "06", "07", "08", "09", "0a", "0b", "0c", "0d", "0e", "0f",
"10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "1a", "1b", "1c", "1d", "1e", "1f",
"20", "21", "22", "23", "24", "25", "26", "27", "28", "29", "2a", "2b", "2c", "2d", "2e", "2f",