@samitouri / QOSami-HFS-2 / commits / 5b3978c4

fix: check for available space on new uploads didn't consider ongoing uploads

Massimo Melina committed Jul 1, 2025 at 00:14 UTC 5b3978c4e4da4ade451bd262f4fbd2ac63cb71ea

3 files changed +37 -18

src/index.ts

@@ -59,6 +59,7 @@ function errorHandler(err:Error & { code:string, path:string }) {

59 || code === 'ERR_STREAM_WRITE_AFTER_END' // happens disconnecting uploads, don't care

60 || code === 'ERR_STREAM_PREMATURE_CLOSE' // happens when many files are sent (not locally), but I checked that the files are written completely. Introduced after node18.5.0 and is thrown by pipeline() used by PUT method handler.

61 || code === 'HPE_INVALID_METHOD' // cannot serve you like that

62 + || code === 'HPE_CLOSED_CONNECTION' // uploads with wrong content-length, but we already handle that properly

63 || code === 'HPE_INVALID_EOF_STATE') return // someone interrupted, don't care

64 console.error('server error', err)

65 }

src/upload.ts

+11 -9

@@ -68,10 +68,9 @@ async function calcHash(fn: string, limit=Infinity) {

68 }

70 const diskSpaceCache = expiringCache<ReturnType<typeof getDiskSpaceSync>>(3_000) // invalidate shortly

71 -const uploadingFiles = new Map<string, Koa.Context>()

71 +const uploadingFiles = new Map<string, { ctx: Koa.Context, size: number, got: number }>()

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 = ''

74 if (dirTraversal(path))

75 return fail(HTTP_FOOL)

76 if (statusCodeForMissingPerm(base, 'can_upload', ctx)) {

@@ -83,7 +82,11 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:

82 return fail()

83 }

84 // enforce minAvailableMb

86 - fullPath = join(base.source!, path)

85 + const fullPath = join(base.source!, path)

86 + const already = uploadingFiles.get(fullPath) // this can be checked so early because this function is sync

87 + 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')

90 const dir = dirname(fullPath)

91 const min = minAvailableMb.get() * (1 << 20)

92 const contentLength = Number(ctx.headers["content-length"])

@@ -105,7 +108,8 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:

108 const { free } = res

109 if (typeof free !== 'number' || isNaN(free))

110 throw ''

108 - if (stillToWrite > free - (min || 0))

111 + const reservedSpace = _.sumBy(Array.from(uploadingFiles.values()), x => x.size - x.got)

112 + if (stillToWrite > free - (min || 0) - reservedSpace)

113 return fail(HTTP_INSUFFICIENT_STORAGE)

114 }

115 catch(e: any) { // warn, but let it through

@@ -114,10 +118,6 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:

118 // optionally 'skip'

119 if (ctx.query.existing === 'skip' && fs.existsSync(fullPath))

120 return fail(HTTP_CONFLICT, 'exists')

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 })

@@ -183,7 +183,8 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:

183 Object.assign(obj, { fileStream })

184 trackProgress()

185 cancelDeletion(tempName)

186 - uploadingFiles.set(fullPath, ctx)

186 + const tracked = { ctx, got: 0, size: stillToWrite }

187 + uploadingFiles.set(fullPath, tracked)

188 console.debug('upload started')

189 // 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))

@@ -274,6 +275,7 @@ export function uploadWriter(base: VfsNode, baseUri: string, path: string, ctx:

275 }

276

277 function checkIfNewUploadBecameLargerThanResumable() {

278 + tracked.got = bytesGot()

279 sendCurrentSize() // keep the client updated in case it needs to resume on disconnection

280 if (isWritingSecondFile && getCurrentSize() > firstResumableStats?.size!)

281 try { // better be sync here, as we don't want the upload to finish in the middle of the rename

tests/test.ts

+25 -9

@@ -1,7 +1,7 @@

1 import test, { describe, before, after } from 'node:test';

2 import { promisify } from 'util'

3 import { srpClientSequence } from '../src/srp'

4 -import { createReadStream, statSync } from 'fs'

4 +import { createReadStream, statfsSync, statSync } from 'fs'

5 import { basename, dirname, resolve } from 'path'

6 import { exec } from 'child_process'

7 import _ from 'lodash'

@@ -238,6 +238,17 @@ describe('after-login', () => {

238 test('delete.method', req(UPLOAD_DEST, 200, { method: 'DELETE' }))

239 test('delete.miss deleted', req(UPLOAD_DEST, 404, { method: 'delete' }))

240 test('upload.too much', reqUpload(UPLOAD_ROOT + 'temp/tooMuch', 400, BIG_CONTENT, BIG_CONTENT.length / 2)) // 400 is caused by nodejs itself, intercepting the mismatch

241 + test('upload.free space', async () => {

242 + const res = statfsSync(ROOT)

243 + const free = res.bavail * res.bsize

244 + const fakeSize = Math.round(free * 0.51)

245 + const r1 = reqUpload(UPLOAD_ROOT + 'temp/free1', 400, makeReadableThatTakes(1000), fakeSize)()

246 + setTimeout(r1.abort, 1500)

247 + await Promise.all([

248 + r1.catch(() => {}),

249 + wait(100).then(() => reqUpload(UPLOAD_ROOT + 'temp/free2', 507, makeReadableThatTakes(500), fakeSize)())

250 + ])

251 + })

252 test('max_dl.account', async () => {

253 const uri = UPLOAD_ROOT + 'temp/big'

254 await reqUpload(uri, 200, BIG_CONTENT)()

@@ -251,7 +262,7 @@ function login(usr: string, pwd=password) {

262 reqApi(cmd, params, (x,res)=> res.statusCode < 400)())

263 }

264

254 -function reqUpload(dest: string, tester: Tester, body?: string | Readable, size?: number, resume?: number) {

265 +function reqUpload(dest: string, tester: Tester, body?: string | Readable, size?: number, resume=0) {

266 if (resume)

267 dest += '?resume=' + resume

268 size ??= (body as any)?.length ?? statSync(SAMPLE_FILE_PATH).size // it's ok that Readable.length is undefined

@@ -270,7 +281,7 @@ function reqUpload(dest: string, tester: Tester, body?: string | Readable, size?

281 }

282 return req(dest, tester, {

283 method: 'PUT',

273 - headers: { 'content-length': size === undefined ? size : size - (resume||0) },

284 + headers: { connection: 'close', 'content-length': size === undefined ? size : size - resume },

285 body: body ?? createReadStream(SAMPLE_FILE_PATH)

286 })

287 }

@@ -313,11 +324,16 @@ const jar = {}

324 function req(url: string, test:Tester, { baseUrl, throttle, ...requestOptions }: XRequestOptions & { throttle?: number, baseUrl?: string }={}) {

325 // passing 'path' keeps it as it is, avoiding internal resolving

326 let abortable // copy abortable interface to returned promise

316 - return () => Object.assign((abortable = httpStream((baseUrl || defaultBaseUrl) + url, { path: url, jar, ...requestOptions })).catch(e => {

317 - if (e.code === 'ECONNREFUSED')

318 - throw e

319 - return e.cause

320 - }).then(process), _.pick(abortable, 'abort'))

327 + return () => Object.assign(

328 + (abortable = httpStream((baseUrl || defaultBaseUrl) + url, { path: url, jar, ...requestOptions }))

329 + .catch(e => {

330 + if (e.code === 'ECONNREFUSED')

331 + throw e

332 + return e.cause

333 + })

334 + .then(process),

335 + _.pick(abortable, 'abort')

336 + )

337

338 async function process(res:any) {

339 //console.debug('sent', requestOptions, 'got', res instanceof Error ? String(res) : [res.status])

@@ -326,7 +342,7 @@ function req(url: string, test:Tester, { baseUrl, throttle, ...requestOptions }:

342 if (typeof test === 'number')

343 test = { status: test }

344 const stream = throttle ? res.pipe(new ThrottledStream(new ThrottleGroup(throttle))) : res

329 - const data = await stream2string(stream)

345 + const data = await stream2string(stream).catch(() => '')

346 const obj = tryJson(data)

347 if (typeof test === 'object') {

348 let { status, mime, re, inList, outList, length, permInList } = test