fix: unreliable upload in case of disconnections
Massimo Melina committed
Jul 10, 2025 at 21:44 UTC
caf56a7febe04cd8e006773d9055d01f004144c5
7 files changed
+131
-242
admin/src/InstalledPlugins.ts
+2
-2
@@ -9,7 +9,7 @@ import {
9
} from '@mui/icons-material'
10
import {
11
CFG, Html, HTTP_FAILED_DEPENDENCY, md, newObj, prefix, with_, xlate, formatTime, formatDate, replaceStringToReact,
12
- callable, tryJson, useAutoScroll
12
+ callable, tryJson, useAutoScroll, NBSP
13
} from './misc'
14
import { alertDialog, confirmDialog, formDialog, toast } from './dialog'
15
import _ from 'lodash'
@@ -238,7 +238,7 @@ export function renderName({ row, value }: any) {
238
: with_(repo?.split('/'), arr => arr?.length !== 2 ? value
239
: h(Fragment, {},
240
h(Link, { href: 'https://github.com/' + repo, target: 'plugin', onClick(ev) { ev.stopPropagation() } }, treatPluginName(arr[1])),
241
- '\xa0by ', arr[0]
241
+ NBSP + 'by ', arr[0]
242
))
243
)
244
frontend/src/upload.ts
+1
-1
@@ -60,7 +60,7 @@ export function showUpload() {
60
uploadState.adding.push(...files.map(f => ({ file: ref(f), path: getFilePath(f), to }))) )
61
},
62
h(FlexV, { className: 'upload-toolbar' },
63
- !props?.can_upload ? t('no_upload_here', "No upload permission for the current folder")
63
+ props && !props.can_upload ? t('no_upload_here', "No upload permission for the current folder")
64
: h(FlexV, {},
65
h(Flex, { center: true, flexWrap: 'wrap', alignItems: 'stretch' },
66
h('button', {
frontend/src/uploadQueue.ts
+46
-112
@@ -1,17 +1,15 @@
1
import {
2
- HTTP_CONFLICT, HTTP_MESSAGES, HTTP_PAYLOAD_TOO_LARGE, HTTP_RANGE_NOT_SATISFIABLE, HTTP_NOT_MODIFIED,
3
- HTTP_INSUFFICIENT_STORAGE, UPLOAD_RESUMABLE, UPLOAD_REQUEST_STATUS, UPLOAD_RESUMABLE_HASH,
4
- buildUrlQueryString, getHFS, pathEncode, pendingPromise, prefix, randomId, tryJson, with_, wait, waitFor,
2
+ HTTP_CONFLICT, HTTP_MESSAGES, HTTP_PAYLOAD_TOO_LARGE, HTTP_RANGE_NOT_SATISFIABLE, HTTP_INSUFFICIENT_STORAGE,
3
+ HTTP_PRECONDITION_FAILED, UPLOAD_TEMP_HASH, MTIME_CHECK,
4
+ buildUrlQueryString, getHFS, pathEncode, pendingPromise, prefix, randomId, tryJson, wait, with_,
5
} from '@hfs/shared'
6
import { state } from './state'
7
-import { getNotifications } from '@hfs/shared/api'
7
import { alertDialog, toast } from './dialog'
8
import { reloadList } from './useFetchList'
9
import { proxy, ref, snapshot, subscribe } from 'valtio'
10
import { createElement as h } from 'react'
11
import _ from 'lodash'
12
import { UploadStatus } from './upload'
14
-import { hfsEvent, onHfsEvent } from './misc'
13
import i18n from './i18n'
14
const { t } = i18n
15
@@ -60,8 +58,10 @@ setInterval(() => {
58
const now = Date.now()
59
const passed = (now - bytesSentTimestamp) / 1000
60
uploadState.speed = bytesSent / passed
63
- if (currentReq && now - stuckSince >= 10_000) // this will normally cause the upload to be retried after long time of no progress
61
+ if (currentReq && now - stuckSince >= 10_000) { // this will normally cause the upload to be retried after long time of no progress
62
currentReq.abort()
63
+ console.debug('upload stuck, aborting')
64
+ }
65
bytesSent = 0 // reset counter
66
bytesSentTimestamp = now
67
@@ -72,13 +72,9 @@ setInterval(() => {
72
}, 2_000)
73
74
let currentReq: XMLHttpRequest | undefined
75
-let overrideStatus = 0
76
-let notificationChannel = ''
75
+const id = randomId()
76
let userAborted = false
78
-let notificationSource: EventSource | undefined
77
let closeLastDialog: undefined | (() => void)
80
-let currentNotificationHandler: (name: string, data: any) => void
81
-const { OPEN } = EventSource
78
79
let reloadOnClose = false
80
export function resetReloadOnClose() {
@@ -87,62 +83,62 @@ export function resetReloadOnClose() {
83
return true
84
}
85
90
-export async function startUpload(toUpload: ToUpload, to: string, startingResume=0) {
91
- const uploadPath = toUpload.path
92
- console.debug('start upload', uploadPath, startingResume)
93
- let resuming = false
94
- let preserveTempFile = undefined
86
+export async function startUpload(toUpload: ToUpload, to: string, resume=0) {
87
+ console.debug('start upload', toUpload.path, resume)
88
uploadState.uploading = toUpload
89
uploadState.progress = 0
90
userAborted = false
98
- await subscribeNotifications() // subscribe before the request
99
- const waitSecondChunk = pendingPromise() // to avoid race condition in case the notification arrives after the first chunk is finished
91
+ let strictResume = true // ask to reject our request if a better resume is available
92
const splitSize = getHFS().splitUploads
93
const fullSize = toUpload.file.size
102
- let offset = startingResume
103
- let resume = startingResume // this will advance with splitSize
104
- let lastWrittenReceived = 0
105
- let stopLooping = false
94
+ const uriPath = to + pathEncode(toUpload.path)
95
+ let stopLooping = false // allow callbacks to stop the loop
96
do { // at least one iteration, even for empty files
107
- offset = Math.max(resume, lastWrittenReceived)
97
const req = currentReq = new XMLHttpRequest()
98
const requestIsOver = pendingPromise()
110
- overrideStatus = 0
99
stuckSince = Date.now()
100
// beware of 'abort' event: it isn't triggered if connection isn't established yet
101
req.onloadend = async () => {
102
try {
103
currentReq = undefined
116
- const status = overrideStatus || req.status
117
- if (status === HTTP_RANGE_NOT_SATISFIABLE) {
118
- lastWrittenReceived = 0
119
- return
120
- }
121
- if (resuming) { // resuming requested
122
- resuming = false // this behavior is only for once, for cancellation of the upload that is in the background while resume is confirmed
123
- stopLooping = true
124
- return
104
+ strictResume = true // reset at each request
105
+ const { status } = req
106
+ if (status === HTTP_RANGE_NOT_SATISFIABLE)
107
+ return stopLooping = true
108
+ if (status === HTTP_PRECONDITION_FAILED) { // resume available
109
+ const size = Number(req.getResponseHeader('x-size'))
110
+ if (req.getResponseHeader(MTIME_CHECK)) { // mtime check not available
111
+ const hashFromServer = fetch(uriPath + '?get=' + UPLOAD_TEMP_HASH).then(r => r.text())
112
+ const hashed = await calcHash(toUpload.file, size) // therefore, we attempt a check using the hash
113
+ if (hashed !== await hashFromServer) {
114
+ strictResume = false
115
+ return console.debug('upload hash mismatch')
116
+ }
117
+ console.debug('upload hash is matching')
118
+ }
119
+ resume = size
120
+ return console.debug('resuming upload', size.toLocaleString())
121
}
122
if (userAborted || status === HTTP_CONFLICT) { // HTTP_CONFLICT = skipped because existing, or upload in progress
123
+ if (req.responseText === 'retry') // it's our previous request that didn't release the lock yet
124
+ return await wait(2000) // wait before resolving `finished`
125
toUpload.error = status ? t('upload_conflict', "already exists") : t`Interrupted` // the I is uppercase because we are just recycling an old string (with all its translations)
126
uploadState.skipped.push(toUpload)
127
}
128
else if (status >= 400)
129
error(status)
132
- else if (!status // request failed at a network level, so try again same file (return), but not too often (wait)
133
- || status === HTTP_NOT_MODIFIED) // our previous request didn't release the lock yet
130
+ else if (!status) // request failed at a network level, so try again same file (return), but not too often (wait)
131
return await wait(2000) // wait before resolving `finished`
132
else {
133
if (splitSize) {
134
resume += splitSize
135
if (resume < fullSize) return // go on with the next chunk
136
}
140
- stopLooping = true
141
- waitSecondChunk.resolve() // we finished, no need to wait
137
uploadState.done.push({ ...toUpload, response: tryJson(req.responseText) })
138
uploadState.doneByte += toUpload!.file.size
139
reloadOnClose = true
140
}
141
+ stopLooping = true
142
requestIsOver.then(workNextFile)
143
}
144
finally {
@@ -151,82 +147,27 @@ export async function startUpload(toUpload: ToUpload, to: string, startingResume
147
}
148
let lastProgress = 0
149
req.upload.onprogress = (e:any) => {
154
- uploadState.partial = e.loaded + offset
150
+ uploadState.partial = e.loaded + resume
151
uploadState.progress = uploadState.partial / fullSize
152
bytesSent += e.loaded - lastProgress
157
- if (e.loaded > lastProgress)
153
+ if (e.loaded > lastProgress) // some progress = not stuck
154
stuckSince = Date.now()
155
lastProgress = e.loaded
156
}
161
- const partial = splitSize && offset + splitSize < fullSize
162
- req.open('PUT', to + pathEncode(uploadPath) + buildUrlQueryString({
163
- notifications: notificationChannel,
164
- giveBack: toUpload.file.lastModified,
165
- ...partial && { partial: fullSize - offset }, // how much space we need
166
- ...offset && { resume: offset, preserveTempFile },
167
- ...toUpload.comment && { comment: toUpload.comment },
168
- ...with_(state.uploadOnExisting, x => x !== 'rename' && { existing: x }), // rename is the default
169
- }), true)
170
- req.send(toUpload.file.slice(offset, splitSize ? offset + splitSize : undefined))
157
+ const partial = splitSize && resume + splitSize < fullSize
158
+ const queryString = buildUrlQueryString({
159
+ id,
160
+ mtime: toUpload.file.lastModified,
161
+ resume: resume + (strictResume ? '!' : ''),
162
+ partial: partial ? fullSize - resume : undefined, // how much space we need
163
+ comment: toUpload.comment || undefined,
164
+ existing: with_(state.uploadOnExisting, x => x !== 'rename' ? x : undefined), // rename is the default
165
+ })
166
+ req.open('PUT', uriPath + queryString, true)
167
+ req.send(toUpload.file.slice(resume, splitSize ? resume + splitSize : undefined))
168
await requestIsOver
172
- if (!startingResume && notificationSource?.readyState === OPEN) // wait only if notifications are currently available
173
- await waitSecondChunk
169
} while (!stopLooping)
170
176
- async function subscribeNotifications() {
177
- // we want to getNotifications once, as establishing a connection is slow in the case of many small files;
178
- // since our handler refers to closures and needs updates, we use indirection via currentNotificationHandler
179
- currentNotificationHandler = notificationHandler
180
- if (notificationChannel) // already subscribed
181
- return waitFor(() => userAborted || notificationSource?.readyState === OPEN) // ensure the system is still alive
182
- notificationChannel = 'upload-' + randomId()
183
- notificationSource = await getNotifications(notificationChannel, async (name, data) => currentNotificationHandler(name, data))
184
- }
185
-
186
- async function notificationHandler(name: string, data: any) {
187
- const {uploading} = uploadState
188
- if (!uploading) return
189
- if (name === UPLOAD_RESUMABLE_HASH)
190
- return hfsEvent(UPLOAD_RESUMABLE_HASH + data.path, data.hash)
191
- if (name === UPLOAD_RESUMABLE) {
192
- waitSecondChunk.resolve()
193
- if (uploadPath !== data.path) return // is it about current file?
194
- if (data.written)
195
- return lastWrittenReceived = data.written
196
- const {size} = data //TODO use toUpload?
197
- if (!size)
198
- return preserveTempFile = undefined
199
- preserveTempFile = true // this is affecting only split-uploads, because is undefined on first chunk (or no chunking)
200
- if (size > toUpload.file.size) return
201
- if (data.giveBack) {
202
- // lastModified doesn't necessarily mean the file has changed, but it seems ok for the time being
203
- if (data.giveBack !== String(toUpload.file.lastModified)) // query params are always string
204
- return console.debug('upload timestamp changed')
205
- console.debug('upload unchanged')
206
- }
207
- else { // timestamp may miss if the file is left by old version, or HFS was killed
208
- const hashFromServer = new Promise<any>(res => onHfsEvent(UPLOAD_RESUMABLE_HASH + uploadPath, res, { once: true }))
209
- const hashed = await calcHash(uploading.file, size) // therefore, we attempt a check using the hash
210
- if (!hashed) return // too late, we are working on another file
211
- if (hashed !== await hashFromServer) return console.debug('upload hash mismatch')
212
- console.debug('upload hash is matching')
213
- }
214
- resuming = true
215
- console.debug('resuming upload', size.toLocaleString())
216
- preserveTempFile = undefined
217
- abortCurrentUpload() // `resuming` will avoid this to be considered skipped
218
- await wait(500) // be sure the server had the time to react to the abort() and unlocked the file, or our next request will fail
219
- return startUpload(toUpload, to, size)
220
- }
221
- if (name === UPLOAD_REQUEST_STATUS) {
222
- if (uploadPath !== data.path) return // is it about current file?
223
- overrideStatus = data.status
224
- if (overrideStatus >= 400)
225
- abortCurrentUpload()
226
- return
227
- }
228
- }
229
-
171
function error(status: number) {
172
const ERRORS = {
173
[HTTP_PAYLOAD_TOO_LARGE]: t`file too large`,
@@ -242,7 +183,6 @@ export async function startUpload(toUpload: ToUpload, to: string, startingResume
183
}
184
185
function workNextFile() {
245
- stopLooping = true
186
uploadState.uploading = undefined
187
uploadState.partial = 0
188
const { qs } = uploadState
@@ -272,11 +212,6 @@ export function abortCurrentUpload(userAskedForIt=false) {
212
}
213
subscribe(uploadState, () => {
214
const [cur] = uploadState.qs
275
- if (!cur?.entries.length) {
276
- notificationChannel = '' // renew channel at each queue for improved security
277
- notificationSource?.close()
278
- return
279
- }
215
if (cur?.entries.length && !uploadState.uploading && !uploadState.paused)
216
void startUpload(cur.entries[0], cur.to)
217
})
@@ -329,7 +264,6 @@ async function calcHash(file: File, limit=Infinity) {
264
const updateUI = _.debounce(() => uploadState.hashing = (limit - left) / limit, 100, { maxWait: 500 })
265
try {
266
while (left > 0) {
332
- if (uploadState.uploading?.file !== file) return // upload aborted
267
const res = await reader.read()
268
if (res.done) break
269
const chunk = res.value.slice(0, left)
src/cross-const.ts
+2
-3
@@ -8,9 +8,8 @@ export const PORT_DISABLED = -1
8
export const NBSP = '\xA0'
9
export const PLUGIN_CUSTOM_REST_PREFIX = '_'
10
export const HFS_REPO = 'rejetto/hfs'
11
-export const UPLOAD_RESUMABLE = 'upload.resumable'
12
-export const UPLOAD_RESUMABLE_HASH = 'upload.hash'
13
-export const UPLOAD_REQUEST_STATUS = 'upload.status'
11
+export const UPLOAD_TEMP_HASH = 'upload-temp-hash'
12
+export const MTIME_CHECK = 'x-mtime-check'
13
export const PREVIOUS_TAG = 'previous'
14
export const ALLOW_SESSION_IP_CHANGE = 'allow_session_ip_change'
15
export const HIDE_IN_TESTS = 'hideInTests' // elements that have variable size, where masking would produce changes, must be hidden
src/serveGuiAndSharedFiles.ts
+31
-10
@@ -5,11 +5,11 @@ import { sendErrorPage } from './errorPages'
5
import events from './events'
6
import {
7
ADMIN_URI, FRONTEND_URI, HTTP_BAD_REQUEST, HTTP_FORBIDDEN, HTTP_METHOD_NOT_ALLOWED, HTTP_NOT_FOUND,
8
- HTTP_UNAUTHORIZED, HTTP_SERVER_ERROR, HTTP_OK, ICONS_URI, HTTP_FAILED_DEPENDENCY
8
+ HTTP_UNAUTHORIZED, HTTP_SERVER_ERROR, HTTP_OK, ICONS_URI, HTTP_FAILED_DEPENDENCY, UPLOAD_TEMP_HASH
9
} from './cross-const'
10
-import { uploadWriter } from './upload'
10
+import { getUploadTempFor, uploadWriter } from './upload'
11
import formidable from 'formidable'
12
-import { Writable } from 'stream'
12
+import { once, Transform, Writable } from 'stream'
13
import { serveFile, serveFileNode } from './serveFile'
14
import { BUILD_TIMESTAMP, DEV, MIME_AUTO, VERSION } from './const'
15
import { zipStreamFromFolder } from './zip'
@@ -17,7 +17,9 @@ import { preventAdminAccess, favicon } from './adminApis'
17
import { serveGuiFiles } from './serveGuiFiles'
18
import mount from 'koa-mount'
19
import { baseUrl } from './listen'
20
-import { asyncGeneratorToReadable, filterMapGenerator, pathEncode, try_ } from './misc'
20
+import { asyncGeneratorToReadable, filterMapGenerator, parseFile, pathEncode, try_ } from './misc'
21
+import XXH from 'xxhashjs'
22
+import fs from 'fs'
23
import { rm } from 'fs/promises'
24
import { setCommentFor } from './comments'
25
import { basicWeb, detectBasicAgent } from './basicWeb'
@@ -53,7 +55,9 @@ export const serveGuiAndSharedFiles: Koa.Middleware = async (ctx, next) => {
55
ctx.state.considerAsGui = true
56
return serveFile(ctx, join(plugin?.folder || '', ICONS_FOLDER, file), MIME_AUTO)
57
}
56
- if (ctx.method === 'PUT') { // curl -T file url/
58
+ const { get } = ctx.query
59
+ const getUploadTempHash = get === UPLOAD_TEMP_HASH
60
+ if (ctx.method === 'PUT' || getUploadTempHash) { // PUT is what you get with `curl -T file url/`
61
const decPath = decodeURIComponent(path)
62
let rest = basename(decPath)
63
const folderUri = pathEncode(dirname(decPath)) // re-encode to get readable urls
@@ -61,6 +65,9 @@ export const serveGuiAndSharedFiles: Koa.Middleware = async (ctx, next) => {
65
if (!folder)
66
return sendErrorPage(ctx, HTTP_NOT_FOUND)
67
ctx.state.uploadPath = decPath
68
+ if (getUploadTempHash)
69
+ return !folder.source ? sendErrorPage(ctx, HTTP_NOT_FOUND)
70
+ : ctx.body = await parseFile(getUploadTempFor(join(folder.source, rest)), calcHash) // negligible memory leak
71
const dest = uploadWriter(folder, folderUri, rest, ctx)
72
if (dest) {
73
ctx.req.pipe(dest).on('error', err => {
@@ -124,7 +131,6 @@ export const serveGuiAndSharedFiles: Koa.Middleware = async (ctx, next) => {
131
return ctx.status = HTTP_SERVER_ERROR
132
}
133
}
127
- const { get } = ctx.query
134
if (node.default && path.endsWith('/') && !get) { // final/ needed on browser to make resource urls correctly with html pages
135
const found = await urlToNode(node.default, ctx, node)
136
if (found && /\.html?/i.test(node.default))
@@ -136,10 +142,10 @@ export const serveGuiAndSharedFiles: Koa.Middleware = async (ctx, next) => {
142
if (!await nodeIsDirectory(node))
143
return node.url ? ctx.redirect(node.url)
144
: !node.source ? sendErrorPage(ctx, HTTP_METHOD_NOT_ALLOWED) // !dir && !source is not supported at this moment
139
- : !statusCodeForMissingPerm(node, 'can_read', ctx) ? serveFileNode(ctx, node) // all good
140
- : ctx.status !== HTTP_UNAUTHORIZED ? null // all errors don't need extra handling, except unauthorized
141
- : detectBasicAgent(ctx) ? (ctx.set('WWW-Authenticate', 'Basic'), sendErrorPage(ctx))
142
- : ctx.query.dl === undefined && (ctx.state.serveApp = true) && serveFrontendFiles(ctx, next)
145
+ : !statusCodeForMissingPerm(node, 'can_read', ctx) ? serveFileNode(ctx, node) // all good
146
+ : ctx.status !== HTTP_UNAUTHORIZED ? null // all errors don't need extra handling, except unauthorized
147
+ : detectBasicAgent(ctx) ? (ctx.set('WWW-Authenticate', 'Basic'), sendErrorPage(ctx))
148
+ : ctx.query.dl === undefined && (ctx.state.serveApp = true) && serveFrontendFiles(ctx, next)
149
if (!path.endsWith('/'))
150
return ctx.redirect(ctx.state.revProxyPath + ctx.originalUrl.replace(/(\?|$)/, '/$1')) // keep query-string, if any
151
if (statusCodeForMissingPerm(node, 'can_list', ctx)) {
@@ -178,6 +184,21 @@ async function sendFolderList(node: VfsNode, ctx: Koa.Context) {
184
}))
185
}
186
187
+async function calcHash(fn: string, limit=Infinity) {
188
+ const hash = XXH.h32()
189
+ const stream = new Transform({
190
+ transform(chunk, enc, done) {
191
+ hash.update(chunk)
192
+ done()
193
+ }
194
+ })
195
+ fs.createReadStream(fn, { end: limit - 1 }).pipe(stream)
196
+ console.debug('hashing', fn)
197
+ await once(stream, 'finish')
198
+ console.debug('hashed', fn)
199
+ return hash.digest().toString(16)
200
+}
201
+
202
declare module "koa" {
203
interface DefaultState {
204
serveApp?: boolean // please, serve the frontend app
src/upload.ts
+47
-110
@@ -1,16 +1,15 @@
1
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, HTTP_NOT_MODIFIED,
4
+ HTTP_CONFLICT, HTTP_FOOL, HTTP_INSUFFICIENT_STORAGE, HTTP_RANGE_NOT_SATISFIABLE, HTTP_BAD_REQUEST, HTTP_NO_CONTENT,
5
+ HTTP_PRECONDITION_FAILED, MTIME_CHECK,
6
} from './const'
7
import { basename, dirname, extname, join } from 'path'
8
import fs from 'fs'
9
import {
10
dirTraversal, loadFileAttr, pendingPromise, storeFileAttr, try_, createStreamLimiter, pathEncode,
11
- enforceFinal, Timeout, with_, parseFile
11
+ enforceFinal, Timeout,
12
} from './misc'
13
-import { notifyClient } from './frontEndApis'
13
import { defineConfig } from './config'
14
import { getDiskSpaceSync } from './util-os'
15
import { disconnect, updateConnection, updateConnectionForCtx } from './connections'
@@ -19,18 +18,19 @@ import { getCurrentUsername } from './auth'
18
import { setCommentFor } from './comments'
19
import _ from 'lodash'
20
import events from './events'
22
-import { rm } from 'fs/promises'
21
+import { rm, rename, utimes } from 'fs/promises'
22
import { expiringCache } from './expiringCache'
23
import { onProcessExit } from './first'
25
-import { once, Transform } from 'stream'
26
-import { utimes } from 'node:fs/promises'
27
-import XXH from 'xxhashjs'
24
25
export const deleteUnfinishedUploadsAfter = defineConfig<undefined|number>('delete_unfinished_uploads_after', 86_400)
26
export const minAvailableMb = defineConfig('min_available_mb', 100)
27
export const dontOverwriteUploading = defineConfig('dont_overwrite_uploading', true)
28
33
-const waitingToBeDeleted: Record<string, { timeout: Timeout, expires: number, mtimeMs?: number, giveBack?: any }> = {}
29
+const waitingToBeDeleted: Record<string, {
30
+ timeout: Timeout, // pending action
31
+ expires: number, // when
32
+ mtime?: any
33
+}> = {}
34
onProcessExit(() => {
35
if (!Object.keys(waitingToBeDeleted).length) return
36
console.log("removing unfinished uploads")
@@ -52,19 +52,8 @@ function setUploadMeta(path: string, ctx: Koa.Context) {
52
})
53
}
54
55
-async function calcHash(fn: string, limit=Infinity) {
56
- const hash = XXH.h32()
57
- const stream = new Transform({
58
- transform(chunk, enc, done) {
59
- hash.update(chunk)
60
- done()
61
- }
62
- })
63
- fs.createReadStream(fn, { end: limit - 1 }).pipe(stream)
64
- console.debug('hashing', fn)
65
- await once(stream, 'finish')
66
- console.debug('hashed', fn)
67
- return hash.digest().toString(16)
55
+export function getUploadTempFor(fullPath: string) {
56
+ return join(dirname(fullPath), 'hfs$upload-' + basename(fullPath).slice(-200))
57
}
58
59
const diskSpaceCache = expiringCache<ReturnType<typeof getDiskSpaceSync>>(3_000) // invalidate shortly
@@ -73,21 +62,14 @@ const uploadingFiles = new Map<string, { ctx: Koa.Context, size: number, got: nu
62
export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx: Koa.Context) {
63
if (dirTraversal(path))
64
return fail(HTTP_FOOL)
76
- if (statusCodeForMissingPerm(base, 'can_upload', ctx)) {
77
- if (!ctx.get('x-hfs-wait')) { // you can disable the following behavior
78
- // avoid waiting hours for just an error
79
- const t = setTimeout(() => disconnect(ctx), 30_000)
80
- ctx.res.on('finish', () => clearTimeout(t))
81
- }
65
+ if (statusCodeForMissingPerm(base, 'can_upload', ctx))
66
return fail()
83
- }
84
- // enforce minAvailableMb
67
const fullPath = join(base.source!, path)
68
const already = uploadingFiles.get(fullPath) // this can be checked so early because this function is sync
69
if (already) // if it's the same client, we tell to retry later
88
- return fail(ctx.query.notifications && ctx.query.notifications === already.ctx.query.notifications ? HTTP_NOT_MODIFIED : HTTP_CONFLICT,
89
- 'already uploading')
70
+ return fail(HTTP_CONFLICT, ctx.query.id && ctx.query.id === already.ctx.query.id ? 'retry' : 'already uploading')
71
const dir = dirname(fullPath)
72
+ // enforce minAvailableMb
73
const min = minAvailableMb.get() * (1 << 20)
74
const contentLength = Number(ctx.headers["content-length"])
75
const isPartial = ctx.query.partial !== undefined // while the presence of "partial" conveys the upload is split...
@@ -102,12 +84,12 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
84
let closestVfsNode = base // if base=root, there's no parent and no original
85
while (closestVfsNode?.parent && !closestVfsNode.original)
86
closestVfsNode = closestVfsNode.parent! // if it's not original, it surely has a parent
105
- const statDir = closestVfsNode!.source!
106
- const res = diskSpaceCache.try(statDir, () => getDiskSpaceSync(statDir))
87
+ const dirToCheck = closestVfsNode!.source!
88
+ const res = diskSpaceCache.try(dirToCheck, () => getDiskSpaceSync(dirToCheck))
89
if (!res) throw 'miss'
90
const { free } = res
91
if (typeof free !== 'number' || isNaN(free))
110
- throw ''
92
+ throw JSON.stringify(res)
93
const reservedSpace = _.sumBy(Array.from(uploadingFiles.values()), x => x.size - x.got)
94
if (stillToWrite > free - (min || 0) - reservedSpace)
95
return fail(HTTP_INSUFFICIENT_STORAGE)
@@ -119,53 +101,33 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
101
if (ctx.query.existing === 'skip' && fs.existsSync(fullPath))
102
return fail(HTTP_CONFLICT, 'exists')
103
let overwriteRequestedButForbidden = false
104
+ const mtime = Number(ctx.query.mtime) || 0
105
try {
123
- const sendCurrentSize = _.debounce(() => notifyClient(ctx, UPLOAD_RESUMABLE, { path, written: getCurrentSize() }), 1000, { maxWait: 1000 })
106
// if upload creates a folder, then add meta to it too
107
if (!dir.endsWith(':\\') && fs.mkdirSync(dir, { recursive: true }))
108
setUploadMeta(dir, ctx)
109
// use temporary name while uploading
128
- const keepName = basename(fullPath).slice(-200)
129
- const firstTempName = join(dir, 'hfs$upload-' + keepName)
130
- const altTempName = join(dir, 'hfs$upload2-' + keepName) // this file makes sense only while smaller than firstTempName
131
- const splitAndPreserving = ctx.query.preserveTempFile // frontend knows about existing temp that can be resumed, but it is not resuming that, but instead it is continuing split-uploading on alternative temp file
132
- let tempName = splitAndPreserving ? altTempName : firstTempName
110
+ const tempName = getUploadTempFor(fullPath)
111
const stats = try_(() => fs.statSync(tempName))
134
- const resumableSize = stats?.size || 0 // we use size to even when user has not required resume, yet, to notify frontend of the possibility
135
- const firstResumableStats = tempName === firstTempName ? stats : try_(() => fs.statSync(firstTempName))
136
- let resumableTempName = resumableSize > 0 ? tempName : undefined
137
- if (resumableTempName)
138
- tempName = altTempName
112
+ const resumableSize = stats?.size || 0
113
// checks for resume feature
140
- let resume = Number(ctx.query.resume)
141
- if (resume > resumableSize)
114
+ const par = String(ctx.query.resume || '')
115
+ const resume = parseInt(par) || 0
116
+ const strictResume = par.at(-1) === '!'
117
+ if (resume > resumableSize || resume < 0)
118
return fail(HTTP_RANGE_NOT_SATISFIABLE)
143
- // warn frontend about resume possibility
144
- let resumeInfo = resumableTempName ? waitingToBeDeleted[resumableTempName] : undefined
145
- if (resumeInfo?.mtimeMs && resumeInfo?.mtimeMs !== try_(() => fs.statSync(resumableTempName!).mtimeMs)) // outdated?
146
- resumeInfo = undefined
147
- if (!resume)
148
- with_(resumableTempName, async x => {
149
- notifyClient(ctx, UPLOAD_RESUMABLE, !x ? { path } : {
150
- path,
151
- size: resumableSize,
152
- // a resumable file exists without a record? then we record it (delayedDelete), plus we provide a hash ASAP, since there's no previous giveBack to compare with
153
- ...resumeInfo || _.omit(delayedDelete(path, deleteUnfinishedUploadsAfter.get() || 0), 'giveBack'), // giveBack makes sense only if coming from resumeObject
154
- timeout: undefined // this entry is here to remove the property copied in the previous line
155
- })
156
- if (x && !resumeInfo)
157
- notifyClient(ctx, UPLOAD_RESUMABLE_HASH, { path, hash: await parseFile(x!, calcHash) }) // negligible memory leak
158
- })
119
+ const resumeInfo = resumableSize && waitingToBeDeleted[tempName]
120
+ if (strictResume) // frontend asked to be notified about resumable uploads
121
+ if (resumableSize > resume && (!resumeInfo || resumeInfo.mtime === mtime)) {
122
+ ctx.set('x-size', String(resumableSize))
123
+ if (!resumeInfo) // if unavailable, the client can request hashing
124
+ ctx.set(MTIME_CHECK, 'not-available')
125
+ return fail(HTTP_PRECONDITION_FAILED)
126
+ }
127
// append if resuming
160
- const resuming = resume && resumableTempName
161
- if (!resuming)
162
- resume = 0
128
+ if (!resume && stats)
129
+ fs.unlinkSync(tempName)
130
const writeStream = createStreamLimiter(contentLength ?? Infinity)
164
- if (resume && resumableTempName && !splitAndPreserving) { // we want to resume the firstTempName, actually
165
- fs.rm(altTempName, () => {})
166
- tempName = resumableTempName
167
- }
168
- let isWritingSecondFile = tempName === altTempName
131
const fullSize = stillToWrite + resume
132
ctx.state.uploadDestinationPath = tempName
133
// allow plugins to mess with the write-stream, because the read-stream can be complicated in case of multipart
@@ -173,8 +135,7 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
135
const resEvent = events.emit('uploadStart', obj)
136
if (resEvent?.isDefaultPrevented()) return
137
176
- const fileStream = resume && resumableTempName ? fs.createWriteStream(resumableTempName, { flags: 'r+', start: resume })
177
- : fs.createWriteStream(tempName)
138
+ const fileStream = fs.createWriteStream(tempName, resume ? { flags: 'r+', start: resume } : undefined)
139
writeStream.on('error', e => {
140
releaseFile()
141
console.debug(e)
@@ -187,27 +148,25 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
148
uploadingFiles.set(fullPath, tracked)
149
console.debug('upload started')
150
// 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
190
- writeStream.on('data', () => setTimeout(checkIfNewUploadBecameLargerThanResumable))
151
+ writeStream.on('data', () => setTimeout(() => tracked.got = bytesGot()))
152
192
- const lockMiddleware = pendingPromise<string>() // expose when all operations stopped
153
+ const lockMiddleware = pendingPromise<string>() // expose outside, to let know when all operations stopped
154
writeStream.once('close', async () => {
155
try {
156
ctx.state.uploadSize = bytesGot() // in case content-length is not specified
157
await new Promise(res => fileStream.close(res)) // this only seems necessary on Windows
158
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
198
- if (isWritingSecondFile) // we don't want to be left with 2 temp files
199
- return rm(altTempName).catch(console.warn)
159
const sec = deleteUnfinishedUploadsAfter.get()
160
return _.isNumber(sec) && delayedDelete(tempName, sec)
161
}
203
- if (isPartial) // we are supposed to leave the upload as unfinished, with the temp name
162
+ if (isPartial) // we are supposed to leave the unfinished upload as it is, with its temp name
163
return ctx.status = HTTP_NO_CONTENT // lockMiddleware contains an empty string, so we must take care of the status
205
- let dest = fullPath
164
+ let dest = fullPath // final destination, considering numbering if necessary
165
if (dontOverwriteUploading.get() && !await overwriteAnyway() && fs.existsSync(dest)) {
166
if (overwriteRequestedButForbidden) {
167
await rm(tempName).catch(console.warn)
168
releaseFile()
210
- return fail()
169
+ return fail() // status code set by overwriteAnyway
170
}
171
const ext = extname(dest)
172
const base = dest.slice(0, -ext.length || Infinity)
@@ -216,15 +175,10 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
175
while (fs.existsSync(dest))
176
}
177
try {
219
- fs.renameSync(tempName, dest) // sync to avoid race conditions with checkIfNewUploadBecameLargerThanResumable
220
- const t = Number(ctx.query.giveBack) // we know giveBack contains lastModified in ms
221
- if (t) // so we use it to touch the file
222
- await utimes(dest, Date.now() / 1000, t / 1000)
178
+ await rename(tempName, dest)
179
+ if (mtime) // so we use it to touch the file
180
+ await utimes(dest, Date.now() / 1000, mtime / 1000)
181
cancelDeletion(tempName) // not necessary, as deletion's failure is silent, but still
224
- if (isWritingSecondFile) { // we've been using altTempName, but now we're done, so we can delete firstTempName
225
- cancelDeletion(firstTempName)
226
- await rm(firstTempName) // wait, so the client can count on the temp-file being gone
227
- }
182
ctx.state.uploadDestinationPath = dest
183
void setUploadMeta(dest, ctx)
184
if (ctx.query.comment)
@@ -265,27 +219,9 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
219
writeStream.once('close', () => clearInterval(h) )
220
}
221
268
- function getCurrentSize() {
269
- return bytesGot() + resume
270
- }
271
-
222
function bytesGot() {
223
return fileStream.bytesWritten + fileStream.writableLength
224
}
275
-
276
- function checkIfNewUploadBecameLargerThanResumable() {
277
- tracked.got = bytesGot()
278
- sendCurrentSize() // keep the client updated in case it needs to resume on disconnection
279
- if (isWritingSecondFile && getCurrentSize() > firstResumableStats?.size!)
280
- try { // better be sync here, as we don't want the upload to finish in the middle of the rename
281
- fs.renameSync(tempName, firstTempName) // try to rename $upload2 to $upload, overwriting
282
- tempName = firstTempName
283
- isWritingSecondFile = false
284
- resumableTempName = undefined
285
- notifyClient(ctx, UPLOAD_RESUMABLE, { path }) // no longer resumable
286
- }
287
- catch{}
288
- }
225
}
226
catch (e: any) {
227
releaseFile()
@@ -303,8 +239,7 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
239
function delayedDelete(path: string, secs: number) {
240
clearTimeout(waitingToBeDeleted[path]?.timeout)
241
return waitingToBeDeleted[path] = {
306
- giveBack: ctx.query.giveBack,
307
- mtimeMs: try_(() => fs.statSync(path).mtimeMs),
242
+ mtime,
243
expires: Date.now() + secs * 1000,
244
timeout: setTimeout(() => {
245
delete waitingToBeDeleted[path]
@@ -327,7 +262,9 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:
262
ctx.status = status
263
if (msg)
264
ctx.body = msg
330
- notifyClient(ctx, UPLOAD_REQUEST_STATUS, { path, status }) // allow browsers to detect failure while still sending body
265
+ if (status >= 400 // with other codes Chrome will report ERR_CONNECTION_RESET
266
+ && !ctx.get('x-hfs-wait')) // you can disable the following behavior
267
+ setTimeout(() => disconnect(ctx), 200) // don't wait, if the upload is still in progress
268
}
269
}
270
tests/test.ts
+2
-4
@@ -203,7 +203,7 @@ describe('after-login', () => {
203
await rm(fn, {force: true})
204
const neededTime = 600
205
const makeAbortedRequest = (afterMs: number) => {
206
- const r = reqUpload(UPLOAD_DEST + '?supposedToAbort', 0, makeReadableThatTakes(neededTime))()
206
+ const r = reqUpload(UPLOAD_DEST + '?supposedToAbort' /*to recognize in the logs*/, 0, makeReadableThatTakes(neededTime))()
207
setTimeout(r.abort, afterMs)
208
return r.catch(() => {}) // wait for it to fail
209
.then(() => wait(500)) // aborted requests don't guarantee that the server has finished and released the file, so we wait some arbitrary time
@@ -214,9 +214,7 @@ describe('after-login', () => {
214
const size = getTempSize()
215
if (!size) // temp file is left, not empty
216
throw Error("missing temp file")
217
- await makeAbortedRequest(timeFirstRequest * .5) // upload less than r1
218
- if (size !== getTempSize()) // shouldn't change, as r2 is smaller, and therefore only wrote to secondary temp file
219
- throw Error(`modified temp file, it was ${size} and now it's ${getTempSize()}`)
217
+ await reqUpload(UPLOAD_DEST + '?resume=0!', 412)()
218
await makeAbortedRequest(timeFirstRequest * 1.5) // upload more than r1
219
if (!(size < getTempSize()!)) // should be increased, as secondary temp file got bigger and replaced primary one
220
throw Error(`temp file not enlarged, it was ${size} and now it's ${getTempSize()}`)