Optimize encryption, storage, and media streaming performance (#447)
- AES-256-GCM in-place decryption halves peak RAM in encrypted blob backend - Offload crypto ≥64 KiB to spawn_blocking (unblocks async runtime) - Fix off-by-one in encrypted range stream (end now exclusive) - Collapse 3 DB round-trips for quota updates into 1 correlated UPDATE - Set-based reconciliation sweep replaces per-user task spawning - Eliminate entity re-read after file overwrite via RETURNING clause - Lightbox streams video/photos inline instead of fetch→blob
This commit is contained in:
Dionisio Pozo
@@ -95,7 +95,10 @@ pub trait BlobStorageBackend: Send + Sync + 'static {
|
||||
/// Stream the full blob content in chunks.
|
||||
fn get_blob_stream(&self, hash: &str) -> BoxFut<'_, Result<BlobStream, DomainError>>;
|
||||
|
||||
/// Stream a byte range of the blob (for HTTP Range requests / video seek).
|
||||
/// Stream the byte range `[start, end)` of the blob (for HTTP Range
|
||||
/// requests / video seek). `end` is **exclusive**; `None` means "to the
|
||||
/// end of the blob". Callers translating inclusive HTTP Range headers
|
||||
/// must pass `last_byte + 1`.
|
||||
fn get_blob_range_stream(
|
||||
&self,
|
||||
hash: &str,
|
||||
|
||||
@@ -286,6 +286,10 @@ pub trait FileWritePort: Send + Sync + 'static {
|
||||
/// When `pre_computed_hash` is provided, the dedup service skips the
|
||||
/// hash re-read — zero extra I/O beyond the initial spool.
|
||||
/// Peak RAM: ~256 KB regardless of file size.
|
||||
///
|
||||
/// Returns `(new_blob_hash, updated_at_epoch)` — everything a caller
|
||||
/// needs to rebuild the fresh entity/ETag from a `File` it already
|
||||
/// holds, without re-reading the row it just updated.
|
||||
async fn update_file_content_from_temp(
|
||||
&self,
|
||||
file_id: &str,
|
||||
@@ -294,7 +298,7 @@ pub trait FileWritePort: Send + Sync + 'static {
|
||||
content_type: Option<String>,
|
||||
pre_computed_hash: Option<String>,
|
||||
modified_at: Option<i64>,
|
||||
) -> Result<String, DomainError>;
|
||||
) -> Result<(String, i64), DomainError>;
|
||||
|
||||
/// Registers file metadata WITHOUT writing content to disk (write-behind).
|
||||
///
|
||||
|
||||
@@ -319,7 +319,8 @@ impl FileUploadUseCase for FileUploadService {
|
||||
&& let Some(file) = file_read.find_file_by_path(path).await?
|
||||
{
|
||||
let file_id = file.id().to_string();
|
||||
self.file_write
|
||||
let (new_hash, updated_at) = self
|
||||
.file_write
|
||||
.update_file_content_from_temp(
|
||||
&file_id,
|
||||
temp_path,
|
||||
@@ -333,8 +334,26 @@ impl FileUploadUseCase for FileUploadService {
|
||||
if let Some(cc) = &self.content_cache {
|
||||
cc.invalidate(&file_id).await;
|
||||
}
|
||||
// Re-read to get fresh DTO with updated etag and timestamps.
|
||||
let updated = file_read.get_file(&file_id).await?;
|
||||
// Rebuild the fresh DTO from the entity already in hand plus the
|
||||
// values the UPDATE just returned — a re-read would only fetch
|
||||
// what we already know, at one extra round-trip per overwrite
|
||||
// (WebDAV sync clients overwrite constantly).
|
||||
let parts = file.into_parts();
|
||||
let updated = crate::domain::entities::file::File::with_timestamps_and_blob_hash(
|
||||
parts.id,
|
||||
parts.name,
|
||||
parts.storage_path,
|
||||
size,
|
||||
parts.mime_type,
|
||||
parts.folder_id,
|
||||
parts.created_at,
|
||||
updated_at as u64,
|
||||
parts.owner_id,
|
||||
new_hash,
|
||||
)
|
||||
.map_err(|e| {
|
||||
DomainError::internal_error("FileUpload", format!("rebuild entity: {e}"))
|
||||
})?;
|
||||
let dto = FileDto::from(updated);
|
||||
if let Some(hook) = &self.file_lifecycle_hook {
|
||||
hook.on_file_updated(&file_id, &dto.etag, content_type);
|
||||
|
||||
@@ -213,8 +213,8 @@ impl FileWritePort for MockFileWritePort {
|
||||
_content_type: Option<String>,
|
||||
_pre_computed_hash: Option<String>,
|
||||
_modified_at: Option<i64>,
|
||||
) -> Result<String, DomainError> {
|
||||
Ok(String::new())
|
||||
) -> Result<(String, i64), DomainError> {
|
||||
Ok((String::new(), 0))
|
||||
}
|
||||
|
||||
async fn register_file_deferred(
|
||||
|
||||
@@ -31,19 +31,36 @@ impl StorageUsageService {
|
||||
}
|
||||
}
|
||||
|
||||
/// Calculates and updates storage usage for a specific user
|
||||
/// Recalculates and stores one user's usage in a single statement.
|
||||
///
|
||||
/// The correlated `SUM(size)` over the user's non-trashed files is
|
||||
/// O(number of files) but runs as an index-only scan on the
|
||||
/// `idx_files_user_size_active` covering partial index. One round-trip
|
||||
/// (was three: user lookup + SUM + UPDATE). NOT called on the request
|
||||
/// path — only by the per-upload background update and the sweep.
|
||||
pub async fn update_user_storage_usage(&self, user_id: Uuid) -> Result<i64, DomainError> {
|
||||
info!("Updating storage usage for user: {}", user_id);
|
||||
let total_usage: Option<i64> = sqlx::query_scalar(
|
||||
r#"
|
||||
UPDATE auth.users u
|
||||
SET storage_used_bytes = COALESCE((
|
||||
SELECT SUM(f.size)::bigint
|
||||
FROM storage.files f
|
||||
WHERE f.user_id = u.id AND NOT f.is_trashed), 0)
|
||||
WHERE u.id = $1
|
||||
RETURNING u.storage_used_bytes
|
||||
"#,
|
||||
)
|
||||
.bind(user_id)
|
||||
.fetch_optional(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| {
|
||||
DomainError::internal_error("StorageUsage", format!("Failed to update usage: {e}"))
|
||||
})?;
|
||||
|
||||
// Calculate storage usage directly from database
|
||||
let total_usage = self.calculate_user_storage_usage(user_id).await?;
|
||||
let total_usage = total_usage
|
||||
.ok_or_else(|| DomainError::not_found("User", format!("User ID: {user_id}")))?;
|
||||
|
||||
// Update the user's storage usage in the database
|
||||
self.user_repository
|
||||
.update_storage_usage(user_id, total_usage)
|
||||
.await?;
|
||||
|
||||
info!(
|
||||
debug!(
|
||||
"Updated storage usage for user {} to {} bytes",
|
||||
user_id, total_usage
|
||||
);
|
||||
@@ -51,61 +68,35 @@ impl StorageUsageService {
|
||||
Ok(total_usage)
|
||||
}
|
||||
|
||||
/// Calculates a user's storage usage by summing all their file sizes.
|
||||
///
|
||||
/// This is `SUM(size)` over the user's non-trashed files — O(number of
|
||||
/// files), backed by the `idx_files_user_size_active` covering partial
|
||||
/// index so it runs as an index-only scan. It is NOT called on the request
|
||||
/// path; only by the per-upload update and the background reconciliation
|
||||
/// sweep.
|
||||
async fn calculate_user_storage_usage(&self, user_id: Uuid) -> Result<i64, DomainError> {
|
||||
debug!("Calculating storage for user: {}", user_id);
|
||||
|
||||
// Direct SQL query to sum all file sizes for this user
|
||||
// This is much more efficient than recursively walking folders
|
||||
let total_size: i64 = sqlx::query_scalar(
|
||||
r#"
|
||||
SELECT COALESCE(SUM(size), 0)::bigint
|
||||
FROM storage.files
|
||||
WHERE user_id = $1 AND NOT is_trashed
|
||||
"#,
|
||||
)
|
||||
.bind(user_id)
|
||||
.fetch_one(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| {
|
||||
DomainError::internal_error("StorageUsage", format!("Failed to calculate usage: {e}"))
|
||||
})?;
|
||||
|
||||
debug!(
|
||||
"Calculated storage for user {}: {} bytes",
|
||||
user_id, total_size
|
||||
);
|
||||
|
||||
Ok(total_size)
|
||||
}
|
||||
|
||||
/// Calculates and updates storage usage for a user identified by username.
|
||||
/// Same as [`Self::update_user_storage_usage`], keyed by username.
|
||||
pub async fn update_user_storage_usage_by_username(
|
||||
&self,
|
||||
username: &str,
|
||||
) -> Result<i64, DomainError> {
|
||||
info!("Updating storage usage for username: {}", username);
|
||||
let total_usage: Option<i64> = sqlx::query_scalar(
|
||||
r#"
|
||||
UPDATE auth.users u
|
||||
SET storage_used_bytes = COALESCE((
|
||||
SELECT SUM(f.size)::bigint
|
||||
FROM storage.files f
|
||||
WHERE f.user_id = u.id AND NOT f.is_trashed), 0)
|
||||
WHERE u.username = $1
|
||||
RETURNING u.storage_used_bytes
|
||||
"#,
|
||||
)
|
||||
.bind(username)
|
||||
.fetch_optional(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| {
|
||||
DomainError::internal_error("StorageUsage", format!("Failed to update usage: {e}"))
|
||||
})?;
|
||||
|
||||
let user = self.user_repository.get_user_by_username(username).await?;
|
||||
let user_id = user.id();
|
||||
let total_usage =
|
||||
total_usage.ok_or_else(|| DomainError::not_found("User", username.to_string()))?;
|
||||
|
||||
// Reuse the existing calculation logic
|
||||
let total_usage = self.calculate_user_storage_usage(user_id).await?;
|
||||
|
||||
// Update the user's storage usage in the database
|
||||
self.user_repository
|
||||
.update_storage_usage(user_id, total_usage)
|
||||
.await?;
|
||||
|
||||
info!(
|
||||
"Updated storage usage for username {} (id={}) to {} bytes",
|
||||
username, user_id, total_usage
|
||||
debug!(
|
||||
"Updated storage usage for username {} to {} bytes",
|
||||
username, total_usage
|
||||
);
|
||||
|
||||
Ok(total_usage)
|
||||
@@ -159,49 +150,49 @@ impl StorageUsagePort for StorageUsageService {
|
||||
StorageUsageService::update_user_storage_usage_by_username(self, username).await
|
||||
}
|
||||
|
||||
/// Reconcile every internal user's cached usage in ONE set-based UPDATE.
|
||||
///
|
||||
/// Replaces the previous shape (paginated user list + one spawned task
|
||||
/// per user, each issuing SUM + UPDATE — up to 2N queries and N
|
||||
/// concurrent tasks fighting for pool connections). A single GROUP BY
|
||||
/// over the covering index feeds all users at once, and the
|
||||
/// `IS DISTINCT FROM` guard skips rewriting rows whose value didn't
|
||||
/// change (no dead-tuple churn for idle users). This also removes the
|
||||
/// old `LIMIT 1000` page cap, which silently left users beyond the
|
||||
/// first thousand unreconciled.
|
||||
///
|
||||
/// External users are excluded — they carry no storage by construction
|
||||
/// (DB CHECK `users_external_no_storage`).
|
||||
async fn update_all_users_storage_usage(&self) -> Result<(), DomainError> {
|
||||
info!("Starting batch update of all users' storage usage");
|
||||
debug!("Starting storage-usage reconciliation sweep");
|
||||
|
||||
// Get the list of all users
|
||||
// include_external=false — external users carry no storage by
|
||||
// construction (DB CHECK `users_external_no_storage`), so there's
|
||||
// nothing to compute for them.
|
||||
let users = self.user_repository.list_users(1000, 0, false).await?;
|
||||
let result = sqlx::query(
|
||||
r#"
|
||||
UPDATE auth.users u
|
||||
SET storage_used_bytes = COALESCE(t.total, 0)
|
||||
FROM auth.users u2
|
||||
LEFT JOIN (
|
||||
SELECT user_id, SUM(size)::bigint AS total
|
||||
FROM storage.files
|
||||
WHERE NOT is_trashed
|
||||
GROUP BY user_id
|
||||
) t ON t.user_id = u2.id
|
||||
WHERE u.id = u2.id
|
||||
AND NOT u2.is_external
|
||||
AND u.storage_used_bytes IS DISTINCT FROM COALESCE(t.total, 0)
|
||||
"#,
|
||||
)
|
||||
.execute(self.pool.as_ref())
|
||||
.await
|
||||
.map_err(|e| {
|
||||
error!("Storage-usage reconciliation sweep failed: {}", e);
|
||||
DomainError::internal_error("StorageUsage", format!("reconciliation sweep: {e}"))
|
||||
})?;
|
||||
|
||||
let mut update_tasks = Vec::new();
|
||||
|
||||
// Process users in parallel
|
||||
for user in users {
|
||||
let user_id = user.id();
|
||||
let service_clone = self.clone();
|
||||
|
||||
// Spawn a background task for each user
|
||||
let task = task::spawn(async move {
|
||||
match service_clone.update_user_storage_usage(user_id).await {
|
||||
Ok(usage) => {
|
||||
debug!(
|
||||
"Updated storage usage for user {}: {} bytes",
|
||||
user_id, usage
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Failed to update storage for user {}: {}", user_id, e);
|
||||
Err(e)
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
update_tasks.push(task);
|
||||
}
|
||||
|
||||
// Wait for all tasks to complete
|
||||
for task in update_tasks {
|
||||
// We don't propagate errors from individual users to avoid failing the entire batch
|
||||
let _ = task.await;
|
||||
}
|
||||
|
||||
info!("Completed batch update of all users' storage usage");
|
||||
info!(
|
||||
"Storage-usage reconciliation corrected {} user(s)",
|
||||
result.rows_affected()
|
||||
);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -594,8 +594,8 @@ impl FileWritePort for MockFileRepository {
|
||||
_content_type: Option<String>,
|
||||
_pre_computed_hash: Option<String>,
|
||||
_modified_at: Option<i64>,
|
||||
) -> std::result::Result<String, DomainError> {
|
||||
Ok(String::new())
|
||||
) -> std::result::Result<(String, i64), DomainError> {
|
||||
Ok((String::new(), 0))
|
||||
}
|
||||
|
||||
async fn register_file_deferred(
|
||||
|
||||
+2
-2
@@ -193,8 +193,8 @@ impl FileWritePort for StubFileWritePort {
|
||||
_content_type: Option<String>,
|
||||
_pre_computed_hash: Option<String>,
|
||||
_modified_at: Option<i64>,
|
||||
) -> Result<String, DomainError> {
|
||||
Ok(String::new())
|
||||
) -> Result<(String, i64), DomainError> {
|
||||
Ok((String::new(), 0))
|
||||
}
|
||||
|
||||
async fn register_file_deferred(
|
||||
|
||||
@@ -137,16 +137,19 @@ impl FileBlobWriteRepository {
|
||||
/// removing the new blob reference.
|
||||
///
|
||||
/// `modified_at`: if `Some`, sets `updated_at` to that Unix timestamp;
|
||||
/// if `None`, uses `NOW()` (server time). Returns the new hash on success.
|
||||
/// if `None`, uses `NOW()` (server time). Returns
|
||||
/// `(new_hash, updated_at_epoch)` on success — the effective timestamp
|
||||
/// is returned so callers can rebuild the fresh entity without
|
||||
/// re-reading the row.
|
||||
async fn swap_blob_hash(
|
||||
&self,
|
||||
file_id: &str,
|
||||
new_hash: &str,
|
||||
new_size: i64,
|
||||
modified_at: Option<i64>,
|
||||
) -> Result<String, DomainError> {
|
||||
) -> Result<(String, i64), DomainError> {
|
||||
// Atomic CTE: capture old hash then update in one round-trip, no TOCTOU.
|
||||
let old_hash = match sqlx::query_scalar::<_, String>(
|
||||
let (old_hash, updated_at) = match sqlx::query_as::<_, (String, i64)>(
|
||||
r#"
|
||||
WITH old AS (
|
||||
SELECT id, blob_hash FROM storage.files WHERE id = $3::uuid FOR UPDATE
|
||||
@@ -156,7 +159,7 @@ impl FileBlobWriteRepository {
|
||||
updated_at = COALESCE(to_timestamp($4), NOW())
|
||||
FROM old
|
||||
WHERE f.id = old.id
|
||||
RETURNING old.blob_hash
|
||||
RETURNING old.blob_hash, EXTRACT(EPOCH FROM f.updated_at)::bigint
|
||||
"#,
|
||||
)
|
||||
.bind(new_hash)
|
||||
@@ -166,7 +169,7 @@ impl FileBlobWriteRepository {
|
||||
.fetch_optional(self.pool.as_ref())
|
||||
.await
|
||||
{
|
||||
Ok(Some(old)) => old,
|
||||
Ok(Some(row)) => row,
|
||||
Ok(None) => {
|
||||
// File not found — compensate: remove the new blob ref
|
||||
if let Err(e) = self.dedup.remove_reference(new_hash).await {
|
||||
@@ -201,7 +204,7 @@ impl FileBlobWriteRepository {
|
||||
);
|
||||
}
|
||||
|
||||
Ok(new_hash.to_string())
|
||||
Ok((new_hash.to_string(), updated_at))
|
||||
}
|
||||
|
||||
/// Like [`FileWritePort::save_file_from_temp`] but also returns whether the
|
||||
@@ -513,7 +516,7 @@ impl FileWritePort for FileBlobWriteRepository {
|
||||
content_type: Option<String>,
|
||||
pre_computed_hash: Option<String>,
|
||||
modified_at: Option<i64>,
|
||||
) -> Result<String, DomainError> {
|
||||
) -> Result<(String, i64), DomainError> {
|
||||
// Streaming: pass pre-computed hash so dedup skips re-reading the file.
|
||||
let dedup_result = self
|
||||
.dedup
|
||||
|
||||
@@ -10,16 +10,31 @@
|
||||
//! dedup still works correctly.
|
||||
//!
|
||||
//! Layout on disk/S3: `[12-byte nonce][ciphertext + 16-byte GCM tag]`
|
||||
//!
|
||||
//! ## Runtime & memory characteristics
|
||||
//!
|
||||
//! GCM is all-or-nothing per blob: a blob can only be decrypted whole, so
|
||||
//! every read materializes the full plaintext. This stays bounded because
|
||||
//! `DedupService` stores all new content as CDC chunks (≤ 1 MiB each) and
|
||||
//! resolves Range requests to the overlapping chunks *before* calling this
|
||||
//! backend — an encrypted seek in a large video decrypts a handful of
|
||||
//! chunks, never the file. The unbounded case is **legacy whole-file
|
||||
//! blobs** written before CDC chunking: a range read of one still decrypts
|
||||
//! the entire blob (re-uploading the file re-stores it chunked).
|
||||
//!
|
||||
//! Crypto work for payloads ≥ 64 KiB runs on the blocking pool so AES-GCM
|
||||
//! never stalls the async runtime, and decryption happens **in place** —
|
||||
//! the ciphertext buffer is reused for the plaintext instead of allocating
|
||||
//! a second copy.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::pin::Pin;
|
||||
|
||||
use aes_gcm::aead::{Aead, KeyInit, OsRng};
|
||||
use aes_gcm::aead::{Aead, AeadInPlace, KeyInit, OsRng};
|
||||
use aes_gcm::{AeadCore, Aes256Gcm, Nonce};
|
||||
use bytes::Bytes;
|
||||
use std::sync::Arc;
|
||||
use tokio::fs;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
use crate::application::ports::blob_storage_ports::{
|
||||
BlobStorageBackend, BlobStream, StorageHealthStatus,
|
||||
@@ -29,6 +44,15 @@ use crate::domain::errors::DomainError;
|
||||
/// Nonce size for AES-256-GCM (96 bits = 12 bytes).
|
||||
const NONCE_SIZE: usize = 12;
|
||||
|
||||
/// Payloads at or above this size run crypto on the blocking pool; below
|
||||
/// it the `spawn_blocking` round-trip costs more than the AES work itself.
|
||||
const CRYPTO_OFFLOAD_THRESHOLD: usize = 64 * 1024;
|
||||
|
||||
/// Emission size for decrypted payloads — matches the 64 KiB chunks the
|
||||
/// unencrypted backends stream, so downstream consumers (HTTP bodies,
|
||||
/// hashers) see the same backpressure shape either way.
|
||||
const PLAINTEXT_EMIT_SIZE: usize = 64 * 1024;
|
||||
|
||||
/// `BlobStorageBackend` decorator that encrypts blobs at rest.
|
||||
pub struct EncryptedBlobBackend {
|
||||
inner: Arc<dyn BlobStorageBackend>,
|
||||
@@ -66,6 +90,51 @@ fn encrypt_bytes(cipher: &Aes256Gcm, data: &[u8]) -> Result<Bytes, DomainError>
|
||||
Ok(Bytes::from(encrypted))
|
||||
}
|
||||
|
||||
/// Decrypt the on-disk layout `[nonce][ciphertext + tag]` **in place**.
|
||||
///
|
||||
/// Consumes the encrypted buffer and reuses it for the plaintext, so peak
|
||||
/// RAM is one buffer — not ciphertext + plaintext side by side (which for
|
||||
/// legacy whole-file blobs would double a multi-hundred-MB allocation).
|
||||
fn decrypt_bytes(cipher: &Aes256Gcm, mut encrypted: Vec<u8>) -> Result<Bytes, DomainError> {
|
||||
if encrypted.len() < NONCE_SIZE {
|
||||
return Err(DomainError::internal_error(
|
||||
"Encryption",
|
||||
"encrypted blob too short (missing nonce)",
|
||||
));
|
||||
}
|
||||
let mut ciphertext = encrypted.split_off(NONCE_SIZE); // `encrypted` keeps the nonce
|
||||
let nonce = Nonce::from_slice(&encrypted);
|
||||
cipher
|
||||
.decrypt_in_place(nonce, b"", &mut ciphertext)
|
||||
.map_err(|e| DomainError::internal_error("Encryption", format!("decrypt failed: {e}")))?;
|
||||
Ok(Bytes::from(ciphertext))
|
||||
}
|
||||
|
||||
/// Run a crypto closure inline for small payloads, on the blocking pool for
|
||||
/// large ones — AES-GCM over megabytes must not stall async workers.
|
||||
async fn offload_crypto<T, F>(work_len: usize, job: F) -> Result<T, DomainError>
|
||||
where
|
||||
T: Send + 'static,
|
||||
F: FnOnce() -> Result<T, DomainError> + Send + 'static,
|
||||
{
|
||||
if work_len < CRYPTO_OFFLOAD_THRESHOLD {
|
||||
return job();
|
||||
}
|
||||
tokio::task::spawn_blocking(job)
|
||||
.await
|
||||
.map_err(|e| DomainError::internal_error("Encryption", format!("crypto task join: {e}")))?
|
||||
}
|
||||
|
||||
/// Turn a decrypted payload into a stream of bounded, zero-copy slices.
|
||||
fn plaintext_stream(data: Bytes) -> BlobStream {
|
||||
let len = data.len();
|
||||
let slices: Vec<Result<Bytes, std::io::Error>> = (0..len)
|
||||
.step_by(PLAINTEXT_EMIT_SIZE)
|
||||
.map(|off| Ok(data.slice(off..len.min(off + PLAINTEXT_EMIT_SIZE))))
|
||||
.collect();
|
||||
Box::pin(futures::stream::iter(slices))
|
||||
}
|
||||
|
||||
impl BlobStorageBackend for EncryptedBlobBackend {
|
||||
fn initialize(
|
||||
&self,
|
||||
@@ -81,7 +150,6 @@ impl BlobStorageBackend for EncryptedBlobBackend {
|
||||
let inner = self.inner.clone();
|
||||
let hash = hash.to_string();
|
||||
let source = source_path.to_path_buf();
|
||||
// Clone cipher key material (Aes256Gcm is not Send-safe to move across await)
|
||||
let cipher = self.cipher.clone();
|
||||
Box::pin(async move {
|
||||
// Read plaintext from source
|
||||
@@ -89,31 +157,14 @@ impl BlobStorageBackend for EncryptedBlobBackend {
|
||||
DomainError::internal_error("Encryption", format!("read source: {e}"))
|
||||
})?;
|
||||
|
||||
// Encrypt: nonce || ciphertext (includes GCM tag)
|
||||
let nonce = Aes256Gcm::generate_nonce(&mut OsRng);
|
||||
let ciphertext = cipher.encrypt(&nonce, plaintext.as_ref()).map_err(|e| {
|
||||
DomainError::internal_error("Encryption", format!("encrypt failed: {e}"))
|
||||
})?;
|
||||
let len = plaintext.len();
|
||||
let encrypted = offload_crypto(len, move || encrypt_bytes(&cipher, &plaintext)).await?;
|
||||
|
||||
// Write encrypted blob to a temp file
|
||||
let tmp = source.with_extension("enc.tmp");
|
||||
let mut file = fs::File::create(&tmp).await.map_err(|e| {
|
||||
DomainError::internal_error("Encryption", format!("create tmp: {e}"))
|
||||
})?;
|
||||
file.write_all(nonce.as_slice()).await.map_err(|e| {
|
||||
DomainError::internal_error("Encryption", format!("write nonce: {e}"))
|
||||
})?;
|
||||
file.write_all(&ciphertext).await.map_err(|e| {
|
||||
DomainError::internal_error("Encryption", format!("write ciphertext: {e}"))
|
||||
})?;
|
||||
file.flush()
|
||||
.await
|
||||
.map_err(|e| DomainError::internal_error("Encryption", format!("flush: {e}")))?;
|
||||
drop(file);
|
||||
|
||||
let result = inner.put_blob(&hash, &tmp).await;
|
||||
let _ = fs::remove_file(&tmp).await;
|
||||
result
|
||||
// Hand the ciphertext straight to the inner backend. The previous
|
||||
// implementation spooled it to a `.enc.tmp` file only for the
|
||||
// inner backend to read it back — a full extra write + read of
|
||||
// every blob that came through this path.
|
||||
inner.put_blob_from_bytes(&hash, encrypted).await
|
||||
})
|
||||
}
|
||||
|
||||
@@ -126,7 +177,8 @@ impl BlobStorageBackend for EncryptedBlobBackend {
|
||||
let hash = hash.to_string();
|
||||
let cipher = self.cipher.clone();
|
||||
Box::pin(async move {
|
||||
let encrypted = encrypt_bytes(&cipher, data.as_ref())?;
|
||||
let encrypted =
|
||||
offload_crypto(data.len(), move || encrypt_bytes(&cipher, data.as_ref())).await?;
|
||||
inner.put_blob_from_bytes(&hash, encrypted).await
|
||||
})
|
||||
}
|
||||
@@ -140,7 +192,8 @@ impl BlobStorageBackend for EncryptedBlobBackend {
|
||||
let hash = hash.to_string();
|
||||
let cipher = self.cipher.clone();
|
||||
Box::pin(async move {
|
||||
let encrypted = encrypt_bytes(&cipher, data.as_ref())?;
|
||||
let encrypted =
|
||||
offload_crypto(data.len(), move || encrypt_bytes(&cipher, data.as_ref())).await?;
|
||||
inner.put_blob_from_bytes_unsynced(&hash, encrypted).await
|
||||
})
|
||||
}
|
||||
@@ -163,28 +216,13 @@ impl BlobStorageBackend for EncryptedBlobBackend {
|
||||
let hash = hash.to_string();
|
||||
let cipher = self.cipher.clone();
|
||||
Box::pin(async move {
|
||||
// Read entire encrypted blob (nonce + ciphertext) into memory for decryption
|
||||
// GCM must see the whole message: collect ciphertext, decrypt in
|
||||
// place off the runtime, then stream zero-copy plaintext slices.
|
||||
let enc_stream = inner.get_blob_stream(&hash).await?;
|
||||
let encrypted = collect_stream(enc_stream).await?;
|
||||
|
||||
if encrypted.len() < NONCE_SIZE {
|
||||
return Err(DomainError::internal_error(
|
||||
"Encryption",
|
||||
"encrypted blob too short (missing nonce)",
|
||||
));
|
||||
}
|
||||
|
||||
let (nonce_bytes, ciphertext) = encrypted.split_at(NONCE_SIZE);
|
||||
let nonce = Nonce::from_slice(nonce_bytes);
|
||||
let plaintext = cipher.decrypt(nonce, ciphertext).map_err(|e| {
|
||||
DomainError::internal_error("Encryption", format!("decrypt failed: {e}"))
|
||||
})?;
|
||||
|
||||
let stream: BlobStream =
|
||||
Box::pin(futures::stream::once(
|
||||
async move { Ok(Bytes::from(plaintext)) },
|
||||
));
|
||||
Ok(stream)
|
||||
let len = encrypted.len();
|
||||
let plaintext = offload_crypto(len, move || decrypt_bytes(&cipher, encrypted)).await?;
|
||||
Ok(plaintext_stream(plaintext))
|
||||
})
|
||||
}
|
||||
|
||||
@@ -199,31 +237,25 @@ impl BlobStorageBackend for EncryptedBlobBackend {
|
||||
let hash = hash.to_string();
|
||||
let cipher = self.cipher.clone();
|
||||
Box::pin(async move {
|
||||
// Must decrypt the full blob then slice the plaintext range
|
||||
// Decrypt the full blob, then slice the plaintext range without
|
||||
// copying. For CDC chunks (every blob written since chunking
|
||||
// landed) this is ≤ 1 MiB; only legacy whole-file blobs pay a
|
||||
// full-blob decrypt here — see the module docs.
|
||||
let enc_stream = inner.get_blob_stream(&hash).await?;
|
||||
let encrypted = collect_stream(enc_stream).await?;
|
||||
let len = encrypted.len();
|
||||
let plaintext = offload_crypto(len, move || decrypt_bytes(&cipher, encrypted)).await?;
|
||||
|
||||
if encrypted.len() < NONCE_SIZE {
|
||||
return Err(DomainError::internal_error(
|
||||
"Encryption",
|
||||
"encrypted blob too short",
|
||||
));
|
||||
}
|
||||
// `end` is exclusive — same contract as `LocalBlobBackend`, whose
|
||||
// implementation reads `end - start` bytes. The previous version
|
||||
// here treated it as inclusive and returned one extra byte on
|
||||
// every bounded range, corrupting 206 responses when encryption
|
||||
// was enabled.
|
||||
let total = plaintext.len();
|
||||
let end_excl = end.map(|e| e as usize).unwrap_or(total).min(total);
|
||||
let start = (start as usize).min(end_excl);
|
||||
|
||||
let (nonce_bytes, ciphertext) = encrypted.split_at(NONCE_SIZE);
|
||||
let nonce = Nonce::from_slice(nonce_bytes);
|
||||
let plaintext = cipher.decrypt(nonce, ciphertext).map_err(|e| {
|
||||
DomainError::internal_error("Encryption", format!("decrypt failed: {e}"))
|
||||
})?;
|
||||
|
||||
let start = start as usize;
|
||||
let end = end.map(|e| (e as usize) + 1).unwrap_or(plaintext.len());
|
||||
let end = end.min(plaintext.len());
|
||||
let start = start.min(end);
|
||||
|
||||
let slice = Bytes::from(plaintext[start..end].to_vec());
|
||||
let stream: BlobStream = Box::pin(futures::stream::once(async move { Ok(slice) }));
|
||||
Ok(stream)
|
||||
Ok(plaintext_stream(plaintext.slice(start..end_excl)))
|
||||
})
|
||||
}
|
||||
|
||||
@@ -326,9 +358,9 @@ mod tests {
|
||||
let decrypted = collect_stream(stream).await.unwrap();
|
||||
assert_eq!(decrypted, data);
|
||||
|
||||
// Read range
|
||||
// Read range — `end` is exclusive, matching LocalBlobBackend
|
||||
let range_stream = encrypted
|
||||
.get_blob_range_stream(hash, 7, Some(15))
|
||||
.get_blob_range_stream(hash, 7, Some(16))
|
||||
.await
|
||||
.unwrap();
|
||||
let range_data = collect_stream(range_stream).await.unwrap();
|
||||
@@ -345,4 +377,103 @@ mod tests {
|
||||
encrypted.delete_blob(hash).await.unwrap();
|
||||
assert!(!encrypted.blob_exists(hash).await.unwrap());
|
||||
}
|
||||
|
||||
/// Payloads above `CRYPTO_OFFLOAD_THRESHOLD` take the spawn_blocking
|
||||
/// path and are emitted as multiple bounded slices — the roundtrip and
|
||||
/// range semantics must be identical to the inline path.
|
||||
#[tokio::test]
|
||||
async fn test_large_blob_offloaded_roundtrip_and_ranges() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs")));
|
||||
local.initialize().await.unwrap();
|
||||
|
||||
let key = EncryptedBlobBackend::generate_key();
|
||||
let encrypted = EncryptedBlobBackend::new(local, &key);
|
||||
|
||||
// 300 KiB of a repeating pattern — crosses the offload threshold and
|
||||
// spans several PLAINTEXT_EMIT_SIZE slices.
|
||||
let data: Vec<u8> = (0..300 * 1024).map(|i| (i % 251) as u8).collect();
|
||||
let hash = "feedbeef1234567890feedbeef1234567890feedbeef1234567890feedbeef12";
|
||||
encrypted
|
||||
.put_blob_from_bytes(hash, Bytes::from(data.clone()))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Full roundtrip
|
||||
let stream = encrypted.get_blob_stream(hash).await.unwrap();
|
||||
let decrypted = collect_stream(stream).await.unwrap();
|
||||
assert_eq!(decrypted, data);
|
||||
|
||||
// Mid-file range crossing an emission boundary (`end` exclusive)
|
||||
let (start, end) = (60_000u64, 200_000u64);
|
||||
let stream = encrypted
|
||||
.get_blob_range_stream(hash, start, Some(end))
|
||||
.await
|
||||
.unwrap();
|
||||
let ranged = collect_stream(stream).await.unwrap();
|
||||
assert_eq!(ranged, &data[start as usize..end as usize]);
|
||||
|
||||
// Open-ended suffix range
|
||||
let stream = encrypted
|
||||
.get_blob_range_stream(hash, 299 * 1024, None)
|
||||
.await
|
||||
.unwrap();
|
||||
let suffix = collect_stream(stream).await.unwrap();
|
||||
assert_eq!(suffix, &data[299 * 1024..]);
|
||||
|
||||
// Range entirely past EOF yields empty content
|
||||
let stream = encrypted
|
||||
.get_blob_range_stream(hash, data.len() as u64 + 10, None)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(collect_stream(stream).await.unwrap().is_empty());
|
||||
|
||||
// Plaintext size reported
|
||||
assert_eq!(encrypted.blob_size(hash).await.unwrap(), data.len() as u64);
|
||||
}
|
||||
|
||||
/// A flipped ciphertext byte must fail GCM authentication, never return
|
||||
/// corrupted plaintext.
|
||||
#[tokio::test]
|
||||
async fn test_tampered_ciphertext_fails_decrypt() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs")));
|
||||
local.initialize().await.unwrap();
|
||||
|
||||
let key = EncryptedBlobBackend::generate_key();
|
||||
let encrypted = EncryptedBlobBackend::new(local.clone(), &key);
|
||||
|
||||
let hash = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef";
|
||||
encrypted
|
||||
.put_blob_from_bytes(hash, Bytes::from_static(b"sensitive payload"))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Corrupt one ciphertext byte on disk (past the 12-byte nonce).
|
||||
let path = local.local_blob_path(hash).expect("local path");
|
||||
let mut raw = std::fs::read(&path).unwrap();
|
||||
raw[NONCE_SIZE] ^= 0xFF;
|
||||
std::fs::write(&path, raw).unwrap();
|
||||
|
||||
assert!(encrypted.get_blob_stream(hash).await.is_err());
|
||||
}
|
||||
|
||||
/// Decrypting with a different key must fail authentication.
|
||||
#[tokio::test]
|
||||
async fn test_wrong_key_fails_decrypt() {
|
||||
let tmp = TempDir::new().unwrap();
|
||||
let local = Arc::new(LocalBlobBackend::new(&tmp.path().join("blobs")));
|
||||
local.initialize().await.unwrap();
|
||||
|
||||
let hash = "aaaabbbbccccddddaaaabbbbccccddddaaaabbbbccccddddaaaabbbbccccdddd";
|
||||
let writer =
|
||||
EncryptedBlobBackend::new(local.clone(), &EncryptedBlobBackend::generate_key());
|
||||
writer
|
||||
.put_blob_from_bytes(hash, Bytes::from_static(b"locked"))
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let reader = EncryptedBlobBackend::new(local, &EncryptedBlobBackend::generate_key());
|
||||
assert!(reader.get_blob_stream(hash).await.is_err());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user