feat(fssync): ensure fs sync once upload is recomposed
This commit is contained in:
@@ -597,10 +597,7 @@ impl ChunkedUploadService {
|
|||||||
return Err(DomainError::new(
|
return Err(DomainError::new(
|
||||||
ErrorKind::InvalidInput,
|
ErrorKind::InvalidInput,
|
||||||
"ChunkedUpload",
|
"ChunkedUpload",
|
||||||
format!(
|
format!("Checksum mismatch: expected {}, got {}", expected, actual),
|
||||||
"Checksum mismatch: expected {}, got {}",
|
|
||||||
expected, actual
|
|
||||||
),
|
|
||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -908,6 +905,22 @@ impl ChunkedUploadService {
|
|||||||
output
|
output
|
||||||
.flush()
|
.flush()
|
||||||
.map_err(|e| format!("Failed to flush assembled file: {e}"))?;
|
.map_err(|e| format!("Failed to flush assembled file: {e}"))?;
|
||||||
|
// ── Durability boundary ────────────────────────────────────
|
||||||
|
// `flush` drains BufWriter's userspace buffer but leaves the
|
||||||
|
// bytes in the kernel page cache. Without `sync_all`, a
|
||||||
|
// power loss between this `complete_upload` returning 2xx
|
||||||
|
// and the OS writeback timer firing (~5 s default) loses
|
||||||
|
// the merged blob — and PG's metadata row references a hash
|
||||||
|
// that no longer exists on disk. Reclaim the BufWriter's
|
||||||
|
// inner File via `into_inner` so we can `sync_all` it; the
|
||||||
|
// BufWriter would otherwise drop without flushing on the
|
||||||
|
// inner handle.
|
||||||
|
let raw_output = output
|
||||||
|
.into_inner()
|
||||||
|
.map_err(|e| format!("into_inner on BufWriter failed: {e}"))?;
|
||||||
|
raw_output
|
||||||
|
.sync_all()
|
||||||
|
.map_err(|e| format!("Failed to fsync assembled file: {e}"))?;
|
||||||
|
|
||||||
// Clean up chunk files (keep assembled) — already on a blocking thread
|
// Clean up chunk files (keep assembled) — already on a blocking thread
|
||||||
for (_index, chunk_path) in &chunks_meta {
|
for (_index, chunk_path) in &chunks_meta {
|
||||||
|
|||||||
@@ -6,7 +6,7 @@
|
|||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
use std::pin::Pin;
|
use std::pin::Pin;
|
||||||
use tokio::fs::{self, File};
|
use tokio::fs::{self, File};
|
||||||
use tokio::io::AsyncSeekExt;
|
use tokio::io::{AsyncSeekExt, AsyncWriteExt};
|
||||||
use tokio_util::io::ReaderStream;
|
use tokio_util::io::ReaderStream;
|
||||||
|
|
||||||
use bytes::Bytes;
|
use bytes::Bytes;
|
||||||
@@ -16,6 +16,53 @@ use crate::application::ports::blob_storage_ports::{
|
|||||||
};
|
};
|
||||||
use crate::domain::errors::{DomainError, ErrorKind};
|
use crate::domain::errors::{DomainError, ErrorKind};
|
||||||
|
|
||||||
|
/// Fsync the directory containing `child_path` so a preceding rename
|
||||||
|
/// or create on `child_path` becomes durable across power loss.
|
||||||
|
///
|
||||||
|
/// On Linux this issues `fsync(2)` on the directory file descriptor —
|
||||||
|
/// the canonical "make the dirent change durable" idiom. macOS does
|
||||||
|
/// the same but only persists to the disk controller (true persistence
|
||||||
|
/// would need `fcntl(F_FULLFSYNC)`, which tokio doesn't expose). On
|
||||||
|
/// Windows, opening a directory needs `FILE_FLAG_BACKUP_SEMANTICS` that
|
||||||
|
/// tokio's `File::open` doesn't set; that platform falls through to
|
||||||
|
/// `Ok(())` after a debug log.
|
||||||
|
///
|
||||||
|
/// Best-effort by design: a failure here is logged but does NOT fail
|
||||||
|
/// the upload, because the blob file itself was just `sync_all`'d and
|
||||||
|
/// is durable on its own. Worst case post-crash recovery: a rename
|
||||||
|
/// "reverts" to the un-renamed name (or stays renamed); the dedup-GC
|
||||||
|
/// cleanup pass handles either side.
|
||||||
|
async fn fsync_parent_dir(child_path: &Path) {
|
||||||
|
let Some(parent) = child_path.parent() else {
|
||||||
|
return;
|
||||||
|
};
|
||||||
|
let parent = parent.to_owned();
|
||||||
|
// std::fs (synchronous) opens directories reliably on Linux/macOS;
|
||||||
|
// do it on the blocking pool so we don't park the tokio worker.
|
||||||
|
let result = tokio::task::spawn_blocking(move || -> std::io::Result<()> {
|
||||||
|
let dir = std::fs::File::open(&parent)?;
|
||||||
|
dir.sync_all()
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
match result {
|
||||||
|
Ok(Ok(())) => {}
|
||||||
|
Ok(Err(e)) => {
|
||||||
|
tracing::warn!(
|
||||||
|
error = %e,
|
||||||
|
path = %child_path.display(),
|
||||||
|
"Blob parent-dir fsync failed (rename durability not guaranteed)"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!(
|
||||||
|
error = %e,
|
||||||
|
path = %child_path.display(),
|
||||||
|
"Blob parent-dir fsync task join failed"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Chunk size for streaming file reads (256 KB).
|
/// Chunk size for streaming file reads (256 KB).
|
||||||
const STREAM_CHUNK_SIZE: usize = 256 * 1024;
|
const STREAM_CHUNK_SIZE: usize = 256 * 1024;
|
||||||
|
|
||||||
@@ -122,15 +169,30 @@ impl BlobStorageBackend for LocalBlobBackend {
|
|||||||
|
|
||||||
// Atomic rename (same filesystem). Falls back to copy+delete for
|
// Atomic rename (same filesystem). Falls back to copy+delete for
|
||||||
// cross-device moves (EXDEV errno 18).
|
// cross-device moves (EXDEV errno 18).
|
||||||
|
//
|
||||||
|
// Durability boundary: the caller is responsible for
|
||||||
|
// having sync_all'd the source file before invoking this
|
||||||
|
// function — both `interfaces::upload_spool::spool_body_to_temp`
|
||||||
|
// and `chunked_upload_service::complete_upload_inner`
|
||||||
|
// (the two production producers of `source_path`) now do
|
||||||
|
// so. We fsync the parent of `blob_path` AFTER the rename
|
||||||
|
// so the dirent change itself becomes durable; without
|
||||||
|
// that, a power loss can resurrect the old (unrenamed)
|
||||||
|
// name even when the file contents survive.
|
||||||
if let Err(e) = fs::rename(&source_path, &blob_path).await {
|
if let Err(e) = fs::rename(&source_path, &blob_path).await {
|
||||||
if e.raw_os_error() == Some(18) {
|
if e.raw_os_error() == Some(18) {
|
||||||
// EXDEV — cross-device link
|
// EXDEV — cross-device link. The copy() target is
|
||||||
|
// a fresh file we created, so fsync it before the
|
||||||
|
// parent-dir fsync below.
|
||||||
fs::copy(&source_path, &blob_path).await.map_err(|ce| {
|
fs::copy(&source_path, &blob_path).await.map_err(|ce| {
|
||||||
DomainError::internal_error(
|
DomainError::internal_error(
|
||||||
"Blob",
|
"Blob",
|
||||||
format!("Failed to copy file to blob store: {}", ce),
|
format!("Failed to copy file to blob store: {}", ce),
|
||||||
)
|
)
|
||||||
})?;
|
})?;
|
||||||
|
if let Ok(f) = fs::File::open(&blob_path).await {
|
||||||
|
let _ = f.sync_all().await;
|
||||||
|
}
|
||||||
let _ = fs::remove_file(&source_path).await;
|
let _ = fs::remove_file(&source_path).await;
|
||||||
} else if fs::try_exists(&blob_path).await.unwrap_or(false) {
|
} else if fs::try_exists(&blob_path).await.unwrap_or(false) {
|
||||||
// Concurrent writer placed the blob — discard our copy
|
// Concurrent writer placed the blob — discard our copy
|
||||||
@@ -144,6 +206,8 @@ impl BlobStorageBackend for LocalBlobBackend {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fsync_parent_dir(&blob_path).await;
|
||||||
|
|
||||||
Ok(file_size)
|
Ok(file_size)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
@@ -163,13 +227,27 @@ impl BlobStorageBackend for LocalBlobBackend {
|
|||||||
return Ok(size);
|
return Ok(size);
|
||||||
}
|
}
|
||||||
|
|
||||||
// Write directly to blob path
|
// Write directly to blob path. `fs::write` is `create +
|
||||||
fs::write(&blob_path, &data).await.map_err(|e| {
|
// write_all + close` — but the close on tokio::fs::File
|
||||||
|
// does NOT fsync, so we open explicitly to keep the
|
||||||
|
// `sync_all` call site obvious. Same durability story as
|
||||||
|
// `put_blob`: the blob file is fsync'd before the parent
|
||||||
|
// directory is, so both the content and the dirent
|
||||||
|
// creation survive a power loss in the same step.
|
||||||
|
let mut file = fs::File::create(&blob_path).await.map_err(|e| {
|
||||||
|
DomainError::internal_error("Blob", format!("Failed to create blob file: {}", e))
|
||||||
|
})?;
|
||||||
|
file.write_all(&data).await.map_err(|e| {
|
||||||
DomainError::internal_error(
|
DomainError::internal_error(
|
||||||
"Blob",
|
"Blob",
|
||||||
format!("Failed to write blob from bytes: {}", e),
|
format!("Failed to write blob from bytes: {}", e),
|
||||||
)
|
)
|
||||||
})?;
|
})?;
|
||||||
|
file.sync_all().await.map_err(|e| {
|
||||||
|
DomainError::internal_error("Blob", format!("Failed to fsync blob file: {}", e))
|
||||||
|
})?;
|
||||||
|
drop(file);
|
||||||
|
fsync_parent_dir(&blob_path).await;
|
||||||
|
|
||||||
Ok(size)
|
Ok(size)
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -140,6 +140,21 @@ impl NextcloudChunkedUploadService {
|
|||||||
out.flush()
|
out.flush()
|
||||||
.await
|
.await
|
||||||
.map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
|
.map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
|
||||||
|
// ── Durability boundary ───────────────────────────────────────
|
||||||
|
// `flush()` only drains tokio's userspace buffer; the bytes
|
||||||
|
// still sit in the kernel page cache until the OS writeback
|
||||||
|
// timer fires (~5 s on default Linux). A power loss in that
|
||||||
|
// window would lose acknowledged data — the metadata row
|
||||||
|
// referencing this blob would survive in PG (synchronous_commit
|
||||||
|
// is on), the blob file would not. `sync_all` issues `fsync(2)`
|
||||||
|
// and only returns once the bytes are on the storage medium
|
||||||
|
// (modulo macOS's well-known `F_FULLFSYNC` caveat — `fsync` there
|
||||||
|
// flushes to the disk controller, not the platter). Cost is
|
||||||
|
// ~1–5 ms on SSD, ~5–50 ms on HDD — invisible against the
|
||||||
|
// assembly I/O we already did.
|
||||||
|
out.sync_all()
|
||||||
|
.await
|
||||||
|
.map_err(|e| DomainError::internal_error("ChunkedUpload", e.to_string()))?;
|
||||||
|
|
||||||
Ok((temp_path, total_size))
|
Ok((temp_path, total_size))
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user