fix: upload: resume attempts are skipped sometimes on bad connections
Massimo Melina committed
May 23, 2025 at 16:41 UTC
ae818ea49112b0ac850dc3039902039880d5d62f
3 files changed
+17
-12
frontend/src/uploadQueue.ts
+5
-4
@@ -1,5 +1,5 @@
1
import {
2
- HTTP_CONFLICT, HTTP_MESSAGES, HTTP_PAYLOAD_TOO_LARGE, HTTP_RANGE_NOT_SATISFIABLE,
2
+ HTTP_CONFLICT, HTTP_MESSAGES, HTTP_PAYLOAD_TOO_LARGE, HTTP_RANGE_NOT_SATISFIABLE, HTTP_NOT_MODIFIED,
3
UPLOAD_RESUMABLE, UPLOAD_REQUEST_STATUS, UPLOAD_RESUMABLE_HASH,
4
buildUrlQueryString, dirname, getHFS, pathEncode, pendingPromise, prefix, randomId, tryJson, with_, wait, waitFor,
5
} from '@hfs/shared'
@@ -127,7 +127,8 @@ export async function startUpload(toUpload: ToUpload, to: string, startingResume
127
}
128
else if (status >= 400)
129
error(status)
130
- else if (!status) // request failed at a network level, so try again same file (return), but not too often (wait)
130
+ else if (!status // request failed at a network level, so try again same file (return), but not too often (wait)
131
+ || status === HTTP_NOT_MODIFIED) // our previous request didn't release the lock yet
132
return await wait(2000) // wait before resolving `finished`
133
else {
134
if (splitSize) {
@@ -160,7 +161,7 @@ export async function startUpload(toUpload: ToUpload, to: string, startingResume
161
uploadPath = prefix('', dirname(uploadPath), '/') + toUpload.name
162
const partial = splitSize && offset + splitSize < fullSize
163
req.open('PUT', to + pathEncode(uploadPath) + buildUrlQueryString({
163
- notificationChannel,
164
+ notifications: notificationChannel,
165
giveBack: toUpload.file.lastModified,
166
...partial && { partial: fullSize - offset }, // how much space we need
167
...offset && { resume: offset, preserveTempFile },
@@ -171,7 +172,7 @@ export async function startUpload(toUpload: ToUpload, to: string, startingResume
172
await requestIsOver
173
if (!startingResume && notificationSource?.readyState === OPEN) // wait only if notifications are currently available
174
await waitSecondChunk
174
- } while (!stopLooping && offset < fullSize)
175
+ } while (!stopLooping)
176
177
async function subscribeNotifications() {
178
// we want to getNotifications once, as establishing a connection is slow in the case of many small files;
src/frontEndApis.ts
+1
-1
@@ -185,7 +185,7 @@ export const frontEndApis: ApiHandlers = {
185
186
export function notifyClient(channel: string | Koa.Context, name: string, data: any) {
187
if (typeof channel !== 'string')
188
- channel = String(channel.query.notificationChannel)
188
+ channel = String(channel.query.notifications)
189
events.emit(NOTIFICATION_PREFIX + channel, name, data)
190
}
191
src/upload.ts
+11
-7
@@ -2,7 +2,7 @@ import { getNodeByName, statusCodeForMissingPerm, VfsNode } from './vfs'
2
import Koa from 'koa'
3
import {
4
HTTP_CONFLICT, HTTP_FOOL, HTTP_INSUFFICIENT_STORAGE, HTTP_RANGE_NOT_SATISFIABLE, HTTP_BAD_REQUEST,
5
- UPLOAD_RESUMABLE, UPLOAD_REQUEST_STATUS, UPLOAD_RESUMABLE_HASH, HTTP_NO_CONTENT,
5
+ UPLOAD_RESUMABLE, UPLOAD_REQUEST_STATUS, UPLOAD_RESUMABLE_HASH, HTTP_NO_CONTENT, HTTP_NOT_MODIFIED,
6
} from './const'
7
import { basename, dirname, extname, join } from 'path'
8
import fs from 'fs'
@@ -68,7 +68,7 @@ async function calcHash(fn: string, limit=Infinity) {
68
}
69
70
const diskSpaceCache = expiringCache<ReturnType<typeof getDiskSpaceSync>>(3_000) // invalidate shortly
71
-const uploadingFiles = new Set()
71
+const uploadingFiles = new Map<string, Koa.Context>()
72
// stay sync because we use this function with formidable()
73
export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx: Koa.Context) {
74
let fullPath = ''
@@ -114,9 +114,10 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
114
// optionally 'skip'
115
if (ctx.query.existing === 'skip' && fs.existsSync(fullPath))
116
return fail(HTTP_CONFLICT, 'exists')
117
- if (uploadingFiles.has(fullPath))
118
- return fail(HTTP_CONFLICT, 'already uploading')
119
- uploadingFiles.add(fullPath)
117
+ const already = uploadingFiles.get(fullPath) // this can be checked so early because this function is sync
118
+ if (already) // if it's the same client, we tell to retry later
119
+ return fail(ctx.query.notifications && ctx.query.notifications === already.query.notifications ? HTTP_NOT_MODIFIED : HTTP_CONFLICT,
120
+ 'already uploading')
121
let overwriteRequestedButForbidden = false
122
try {
123
const sendCurrentSize = _.debounce(() => notifyClient(ctx, UPLOAD_RESUMABLE, { path, written: getCurrentSize() }), 1000, { maxWait: 1000 })
@@ -182,6 +183,8 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
183
writeStream.pipe(fileStream)
184
Object.assign(obj, { fileStream })
185
trackProgress()
186
+ uploadingFiles.set(fullPath, ctx)
187
+ console.debug('upload started')
188
// the file stream doesn't have an event for data being written, so we use 'data' of its feeder, which happens before, so we postpone a bit, trying to have a fresher number
189
writeStream.on('data', () => setTimeout(checkIfNewUploadBecameLargerThanResumable))
190
@@ -190,7 +193,7 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
193
try {
194
ctx.state.uploadSize = bytesGot() // in case content-length is not specified
195
await new Promise(res => fileStream.close(res)) // this only seems necessary on Windows
193
- if (ctx.isAborted()) {
196
+ if (ctx.isAborted()) { // in the very unlikely case the connection is interrupted between last-byte and here, we still consider it unfinished, as the client had no way to know, and will resume, but it would get an error if we finish the process
197
if (isWritingSecondFile) // we don't want to be left with 2 temp files
198
return rm(altTempName).catch(console.warn)
199
const sec = deleteUnfinishedUploadsAfter.get()
@@ -202,6 +205,7 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
205
if (dontOverwriteUploading.get() && !await overwriteAnyway() && fs.existsSync(dest)) {
206
if (overwriteRequestedButForbidden) {
207
await rm(tempName).catch(console.warn)
208
+ releaseFile()
209
return fail()
210
}
211
const ext = extname(dest)
@@ -220,6 +224,7 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
224
cancelDeletion(firstTempName)
225
await rm(firstTempName) // wait, so the client can count on the temp-file being gone
226
}
227
+ releaseFile()
228
ctx.state.uploadDestinationPath = dest
229
void setUploadMeta(dest, ctx)
230
if (ctx.query.comment)
@@ -319,7 +324,6 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
324
325
function fail(status=ctx.status, msg?: string) {
326
console.debug('upload failed', status, msg||'')
322
- releaseFile()
327
ctx.status = status
328
if (msg)
329
ctx.body = msg