feat(upload): idempotent re-upload + auto-retry so partial folders self-complete
Re-uploading a partially-uploaded folder used to surface hundreds of spurious "already exists" failures, and a file the watchdog aborted (or one the server committed just before the client gave up) was lost. Backend — save_file_with_blob_impl (the shared write path for both plain and by-hash uploads): on a name conflict (23505), if the existing non-trashed file holds byte-identical content (same folder, same name, same blob hash), return that file as success instead of erroring. A different-content clash still conflicts. Re-upload / re-sync becomes a clean no-op for everything already stored — only the genuinely missing files transfer. Frontend — uploadWithRetry: each file gets one automatic retry on a transient failure (quota is never retried). With backend idempotency, retrying an already-stored file is an instant no-op and a stalled/aborted file gets a real second chance, so a folder upload self-completes instead of leaving gaps. cargo test: 448 passed. npm run check: clean, 58 tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -348,12 +348,35 @@
|
|||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Upload one file, retrying once on a transient failure. A connection the
|
||||||
|
* watchdog aborted, or an upload the server actually committed before the
|
||||||
|
* client gave up, both recover here: the backend treats a re-upload of
|
||||||
|
* byte-identical content as success (idempotent), so the retry is a clean
|
||||||
|
* no-op for anything that already landed and a real second chance for the
|
||||||
|
* rest. A quota error is never retried — the disk won't free up mid-batch.
|
||||||
|
*/
|
||||||
|
async function uploadWithRetry(
|
||||||
|
folderId: string | null,
|
||||||
|
file: File,
|
||||||
|
report: (frac: number) => void,
|
||||||
|
ownedHash: string | null
|
||||||
|
): Promise<number> {
|
||||||
|
try {
|
||||||
|
return await withTimeout(uploadOneFile(folderId, file, report, ownedHash), FILE_BACKSTOP_MS);
|
||||||
|
} catch (e) {
|
||||||
|
if ((e as { isQuota?: boolean } | null)?.isQuota) throw e;
|
||||||
|
report(0); // reset this file's progress for the second attempt
|
||||||
|
return await withTimeout(uploadOneFile(folderId, file, report, ownedHash), FILE_BACKSTOP_MS);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Upload `items` ({file, folderId}) with bounded concurrency, a per-file
|
* Upload `items` ({file, folderId}) with bounded concurrency, a per-file
|
||||||
* deadline and live aggregate progress. A stuck or failing file no longer
|
* deadline and live aggregate progress. A stuck or failing file no longer
|
||||||
* freezes the batch: it blocks only its own lane (the rest keep going) and
|
* freezes the batch: it blocks only its own lane (the rest keep going), is
|
||||||
* eventually times out / is skipped. Quota exhaustion stops the run early.
|
* retried once, and only then counted as failed. Quota exhaustion stops the
|
||||||
* Returns the bytes deduplicated and the count of files that failed.
|
* run early. Returns the bytes deduplicated and the count of files that failed.
|
||||||
*/
|
*/
|
||||||
async function uploadAll(
|
async function uploadAll(
|
||||||
items: { file: File; folderId: string | null }[],
|
items: { file: File; folderId: string | null }[],
|
||||||
@@ -377,19 +400,12 @@
|
|||||||
while (next < total) {
|
while (next < total) {
|
||||||
const i = next++;
|
const i = next++;
|
||||||
const { file, folderId } = items[i];
|
const { file, folderId } = items[i];
|
||||||
|
const report = (f: number) => {
|
||||||
|
if (!Number.isNaN(f)) frac[i] = Math.min(1, f);
|
||||||
|
refresh();
|
||||||
|
};
|
||||||
try {
|
try {
|
||||||
savedBytes += await withTimeout(
|
savedBytes += await uploadWithRetry(folderId, file, report, owned.get(file) ?? null);
|
||||||
uploadOneFile(
|
|
||||||
folderId,
|
|
||||||
file,
|
|
||||||
(f) => {
|
|
||||||
if (!Number.isNaN(f)) frac[i] = Math.min(1, f);
|
|
||||||
refresh();
|
|
||||||
},
|
|
||||||
owned.get(file) ?? null
|
|
||||||
),
|
|
||||||
FILE_BACKSTOP_MS
|
|
||||||
);
|
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
failures++;
|
failures++;
|
||||||
// A full disk won't recover within this batch — stop pulling new
|
// A full disk won't recover within this batch — stop pulling new
|
||||||
|
|||||||
@@ -355,6 +355,36 @@ impl FileBlobWriteRepository {
|
|||||||
if let sqlx::Error::Database(ref db_err) = e
|
if let sqlx::Error::Database(ref db_err) = e
|
||||||
&& db_err.code().as_deref() == Some("23505")
|
&& db_err.code().as_deref() == Some("23505")
|
||||||
{
|
{
|
||||||
|
// Idempotent re-upload: if the conflicting file already
|
||||||
|
// holds IDENTICAL content (same folder, same name, same
|
||||||
|
// blob hash), treat this as success and return that file
|
||||||
|
// instead of erroring. Re-uploading a partially-uploaded
|
||||||
|
// folder then becomes a clean no-op for everything that
|
||||||
|
// already landed — only the genuinely missing files
|
||||||
|
// transfer — instead of surfacing hundreds of spurious
|
||||||
|
// "already exists" failures. The duplicate blob reference
|
||||||
|
// taken during ingest was just released above, so the
|
||||||
|
// existing file's own reference is the only one (correct);
|
||||||
|
// a different-content clash still returns the conflict.
|
||||||
|
match self.fetch_identical_file(fid, &name, blob_hash).await {
|
||||||
|
Ok(Some(existing)) => {
|
||||||
|
tracing::info!(
|
||||||
|
"♻️ IDEMPOTENT UPLOAD: {} already present, identical content (hash: {})",
|
||||||
|
name,
|
||||||
|
&blob_hash[..12]
|
||||||
|
);
|
||||||
|
return Ok(existing);
|
||||||
|
}
|
||||||
|
Ok(None) => {} // genuine conflict (different content)
|
||||||
|
Err(lookup_err) => {
|
||||||
|
tracing::warn!(
|
||||||
|
"idempotency lookup failed for {} (hash {}): {} — returning conflict",
|
||||||
|
name,
|
||||||
|
&blob_hash[..12],
|
||||||
|
lookup_err
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
return Err(DomainError::already_exists(
|
return Err(DomainError::already_exists(
|
||||||
"File",
|
"File",
|
||||||
format!("'{name}' already exists in this folder"),
|
format!("'{name}' already exists in this folder"),
|
||||||
@@ -389,6 +419,85 @@ impl FileBlobWriteRepository {
|
|||||||
updated_by,
|
updated_by,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Fetch a non-trashed file in `folder_id` named `name` whose content blob
|
||||||
|
/// is `blob_hash` — the "is this re-upload byte-identical?" probe that makes
|
||||||
|
/// uploads idempotent on a name conflict. `Ok(None)` means the conflicting
|
||||||
|
/// file has *different* content (a genuine clash the caller must report).
|
||||||
|
async fn fetch_identical_file(
|
||||||
|
&self,
|
||||||
|
folder_id: &str,
|
||||||
|
name: &str,
|
||||||
|
blob_hash: &str,
|
||||||
|
) -> Result<Option<File>, DomainError> {
|
||||||
|
let row = sqlx::query_as::<
|
||||||
|
_,
|
||||||
|
(
|
||||||
|
String,
|
||||||
|
Uuid,
|
||||||
|
String,
|
||||||
|
i64,
|
||||||
|
i64,
|
||||||
|
Option<Uuid>,
|
||||||
|
Option<Uuid>,
|
||||||
|
i64,
|
||||||
|
String,
|
||||||
|
),
|
||||||
|
>(
|
||||||
|
r#"
|
||||||
|
SELECT f.id::text, f.user_id, fo.path,
|
||||||
|
EXTRACT(EPOCH FROM f.created_at)::bigint,
|
||||||
|
EXTRACT(EPOCH FROM f.updated_at)::bigint,
|
||||||
|
f.created_by, f.updated_by, f.size, f.mime_type
|
||||||
|
FROM storage.files f
|
||||||
|
JOIN storage.folders fo ON fo.id = f.folder_id
|
||||||
|
WHERE f.folder_id = $1::uuid
|
||||||
|
AND f.name = $2
|
||||||
|
AND f.blob_hash = $3
|
||||||
|
AND NOT f.is_trashed
|
||||||
|
LIMIT 1
|
||||||
|
"#,
|
||||||
|
)
|
||||||
|
.bind(folder_id)
|
||||||
|
.bind(name)
|
||||||
|
.bind(blob_hash)
|
||||||
|
.fetch_optional(self.pool.as_ref())
|
||||||
|
.await
|
||||||
|
.map_err(|e| {
|
||||||
|
DomainError::internal_error("FileBlobWrite", format!("idempotency lookup: {e}"))
|
||||||
|
})?;
|
||||||
|
|
||||||
|
let Some((
|
||||||
|
id,
|
||||||
|
user_id,
|
||||||
|
folder_path,
|
||||||
|
created_at,
|
||||||
|
updated_at,
|
||||||
|
created_by,
|
||||||
|
updated_by,
|
||||||
|
size,
|
||||||
|
mime_type,
|
||||||
|
)) = row
|
||||||
|
else {
|
||||||
|
return Ok(None);
|
||||||
|
};
|
||||||
|
|
||||||
|
Self::row_to_file(
|
||||||
|
id,
|
||||||
|
name.to_string(),
|
||||||
|
Some(folder_id.to_string()),
|
||||||
|
Some(folder_path),
|
||||||
|
size,
|
||||||
|
mime_type,
|
||||||
|
created_at,
|
||||||
|
updated_at,
|
||||||
|
Some(user_id),
|
||||||
|
blob_hash.to_string(),
|
||||||
|
created_by,
|
||||||
|
updated_by,
|
||||||
|
)
|
||||||
|
.map(Some)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
impl FileWritePort for FileBlobWriteRepository {
|
impl FileWritePort for FileBlobWriteRepository {
|
||||||
|
|||||||
Reference in New Issue
Block a user