test: interrupted upload
Massimo Melina committed
Jan 25, 2025 at 17:22 UTC
62f1a7a79de33c1209a9d865ef1680872a659ad4
2 files changed
+42
-11
src/util-http.ts
+6
-2
@@ -31,8 +31,10 @@ export interface XRequestOptions extends https.RequestOptions {
31
}
32
33
export declare namespace httpStream { let defaultProxy: string | undefined }
34
-export function httpStream(url: string, { body, jar, noRedirect, httpThrow, proxy, ...options }: XRequestOptions ={}): Promise<IncomingMessage> {
35
- return new Promise((resolve, reject) => {
34
+export function httpStream(url: string, { body, jar, noRedirect, httpThrow, proxy, ...options }: XRequestOptions ={}) {
35
+ const controller = new AbortController()
36
+ options.signal ??= controller.signal
37
+ return Object.assign(new Promise<IncomingMessage>((resolve, reject) => {
38
proxy ??= httpStream.defaultProxy
39
options.headers ??= {}
40
if (body) {
@@ -81,6 +83,8 @@ export function httpStream(url: string, { body, jar, noRedirect, httpThrow, prox
83
body.pipe(req).on('end', () => req.end())
84
else
85
req.end(body)
86
+ }), {
87
+ abort() { controller.abort() }
88
})
89
}
90
tests/test.ts
+36
-9
@@ -1,9 +1,9 @@
1
import { srpClientSequence } from '../src/srp'
2
import { createReadStream, statSync } from 'fs'
3
-import { basename, dirname, join, resolve } from 'path'
3
+import { basename, dirname, resolve } from 'path'
4
import { exec } from 'child_process'
5
import _ from 'lodash'
6
-import { findDefined, randomId, tryJson, wait } from '../src/cross'
6
+import { findDefined, randomId, try_, tryJson, wait } from '../src/cross'
7
import { httpStream, stream2string, XRequestOptions } from '../src/util-http'
8
import { ThrottledStream, ThrottleGroup } from '../src/ThrottledStream'
9
import { rm, writeFile } from 'fs/promises'
@@ -38,6 +38,11 @@ class StringRepeaterStream extends Readable {
38
}
39
}
40
41
+function makeReadableThatTakes(ms: number) {
42
+ return Object.assign(Readable.from(BIG_CONTENT).pipe(new ThrottledStream(new ThrottleGroup(BIG_CONTENT.length / ms))),
43
+ { length: BIG_CONTENT.length })
44
+}
45
+
46
describe('basics', () => {
47
//before(async () => appStarted)
48
it('frontend', req('/', /<body>/, { headers: { accept: '*/*' } })) // workaround: 'accept' is necessary when running server-for-test-dev, still don't know why
@@ -174,10 +179,9 @@ describe('after-login', () => {
179
it('upload.ok', reqUpload(UPLOAD_DEST, 200))
180
it('upload.crossing', reqUpload(UPLOAD_DEST.replace('temp', '../..'), 418))
181
it('upload.overlap', async () => {
177
- const seconds = .3
178
- const throttled = Readable.from(BIG_CONTENT).pipe(new ThrottledStream(new ThrottleGroup(BIG_CONTENT.length / 1000 / seconds)))
179
- const first = reqUpload(UPLOAD_DEST, 200, throttled, BIG_CONTENT.length)()
180
- await wait(100)
182
+ const ms = 300
183
+ const first = reqUpload(UPLOAD_DEST, 200, makeReadableThatTakes(ms))()
184
+ await wait(ms/3)
185
await reqUpload(UPLOAD_DEST, 409)() // should conflict
186
await first
187
})
@@ -185,6 +189,28 @@ describe('after-login', () => {
189
reqUpload(UPLOAD_DEST, 200, new StringRepeaterStream(BIG_CONTENT, 150))(), // 300MB
190
..._.range(3).map(i => reqUpload(UPLOAD_DEST + i, 200, new StringRepeaterStream(BIG_CONTENT, 50))()) // 3 x 100MB
191
])).timeout(5000)
192
+ it('upload.interrupted', async () => {
193
+ const fn = resolve(__dirname, UPLOAD_RELATIVE.replace('/', '/hfs$upload-'))
194
+ await rm(fn, {force: true})
195
+ const neededTime = 300
196
+ const makeAbortedRequest = (afterMs: number) => {
197
+ const r = reqUpload(UPLOAD_DEST, 0, makeReadableThatTakes(neededTime))()
198
+ setTimeout(r.abort, afterMs)
199
+ return r.catch(() => {}) // wait for it to fail
200
+ }
201
+ const timeFirstRequest = neededTime * .5 // not enough to finish
202
+ await makeAbortedRequest(timeFirstRequest)
203
+ const getTempSize = () => try_(() => statSync(fn)?.size)
204
+ const size = getTempSize()
205
+ if (!size) // shouldn't be empty
206
+ throw Error("missing temp file")
207
+ await makeAbortedRequest(timeFirstRequest * .5) // upload less than r1
208
+ if (size !== getTempSize()) // shouldn't change, as r2 is smaller
209
+ throw Error("modified temp file")
210
+ await reqUpload(UPLOAD_DEST, 200, makeReadableThatTakes(0))() // quickly complete the upload, and check for final size
211
+ if (getTempSize())
212
+ throw Error("temp file should be cleared")
213
+ })
214
const renameTo = 'z'
215
it('rename.ok', reqApi('rename', { uri: UPLOAD_DEST, dest: renameTo }, 200))
216
it('delete.miss renamed', reqApi('delete', { uri: UPLOAD_DEST }, 404))
@@ -212,7 +238,7 @@ describe('after-login', () => {
238
await reqUpload(uri, 200, BIG_CONTENT)()
239
await testMaxDl(uri, 2, 1)
240
})
215
- after(() => rm(join(__dirname, 'temp'), { recursive: true }).catch(() => 0))
241
+ after(() => rm(resolve(__dirname, 'temp'), { recursive: true }).catch(() => 0))
242
})
243
244
function login(usr: string, pwd=password) {
@@ -273,11 +299,12 @@ const jar = {}
299
300
function req(url: string, test:Tester, { baseUrl, throttle, ...requestOptions }: XRequestOptions & { throttle?: number, baseUrl?: string }={}) {
301
// passing 'path' keeps it as it is, avoiding internal resolving
276
- return () => httpStream((baseUrl || defaultBaseUrl) + url, { path: url, jar, ...requestOptions }).catch(e => {
302
+ let abortable // copy abortable interface to returned promise
303
+ return () => Object.assign((abortable = httpStream((baseUrl || defaultBaseUrl) + url, { path: url, jar, ...requestOptions })).catch(e => {
304
if (e.code === 'ECONNREFUSED')
305
throw e
306
return e.cause
280
- }).then(process)
307
+ }).then(process), _.pick(abortable, 'abort'))
308
309
async function process(res:any) {
310
//console.debug('sent', requestOptions, 'got', res instanceof Error ? String(res) : [res.status])