From e6ee5988ab37e68ee1237bb1e3952898eae91d6a Mon Sep 17 00:00:00 2001 From: DioCrafts Date: Sat, 20 Jun 2026 18:56:05 +0200 Subject: [PATCH] feat(upload): idempotent re-upload + auto-retry so partial folders self-complete MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../src/routes/files/[...path]/+page.svelte | 46 +++++--- .../pg/file_blob_write_repository.rs | 109 ++++++++++++++++++ 2 files changed, 140 insertions(+), 15 deletions(-) diff --git a/frontend/src/routes/files/[...path]/+page.svelte b/frontend/src/routes/files/[...path]/+page.svelte index d1566cd9..169bca9a 100644 --- a/frontend/src/routes/files/[...path]/+page.svelte +++ b/frontend/src/routes/files/[...path]/+page.svelte @@ -348,12 +348,35 @@ 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 { + 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 * 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 - * eventually times out / is skipped. Quota exhaustion stops the run early. - * Returns the bytes deduplicated and the count of files that failed. + * freezes the batch: it blocks only its own lane (the rest keep going), is + * retried once, and only then counted as failed. Quota exhaustion stops the + * run early. Returns the bytes deduplicated and the count of files that failed. */ async function uploadAll( items: { file: File; folderId: string | null }[], @@ -377,19 +400,12 @@ while (next < total) { const i = next++; const { file, folderId } = items[i]; + const report = (f: number) => { + if (!Number.isNaN(f)) frac[i] = Math.min(1, f); + refresh(); + }; try { - savedBytes += await withTimeout( - uploadOneFile( - folderId, - file, - (f) => { - if (!Number.isNaN(f)) frac[i] = Math.min(1, f); - refresh(); - }, - owned.get(file) ?? null - ), - FILE_BACKSTOP_MS - ); + savedBytes += await uploadWithRetry(folderId, file, report, owned.get(file) ?? null); } catch (e) { failures++; // A full disk won't recover within this batch — stop pulling new diff --git a/src/infrastructure/repositories/pg/file_blob_write_repository.rs b/src/infrastructure/repositories/pg/file_blob_write_repository.rs index 3bb60761..cd258f8b 100644 --- a/src/infrastructure/repositories/pg/file_blob_write_repository.rs +++ b/src/infrastructure/repositories/pg/file_blob_write_repository.rs @@ -355,6 +355,36 @@ impl FileBlobWriteRepository { if let sqlx::Error::Database(ref db_err) = e && 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( "File", format!("'{name}' already exists in this folder"), @@ -389,6 +419,85 @@ impl FileBlobWriteRepository { 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, DomainError> { + let row = sqlx::query_as::< + _, + ( + String, + Uuid, + String, + i64, + i64, + Option, + Option, + 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 {