tus-js-client: abort() followed by start() is broken on non-file streams

Describe the bug With a non-file stream, when an upload is start()-ed and then after a few seconds of progress, we call abort(), followed by another start() a few seconds later, the upload grinds to a halt and tus-js-client leaks stream event listeners. This does not happen with file based streams (fs.createReadStream).

To Reproduce

mkdir stream-issue
cd stream-issue
yarn add tus-js-client into-stream@6 throttle

create index.js:

const fs = require('fs')
const { Upload } = require('tus-js-client');
const intoStream = require('into-stream');
const Throttle = require('throttle');

/*
const { Server, FileStore } = require('tus-node-server');
const server = new Server();
server.datastore = new FileStore({
    path: '/files'
});

const host = '127.0.0.1';
const port = 1080;
server.listen({ host, port }, () => {
    console.log(`[${new Date().toLocaleTimeString()}] tus server listening at http://${host}:${port}`);
}); */


// Replace this with the URL of an assembly that is in the state "UPLOADING", for example by pausing an upload using uppy
const assemblyUrl = 'http://api2.qarq.transloadit.com/assemblies/5c990a3f19cd48e8b457d20b8bea11a1'


;(async () => {
  const buf = Buffer.alloc(3e6) // 3MB
  const uploadSize = buf.length


  // THIS CODE REPRODUCES THE PROBLEM:
  const source = intoStream(buf)
  const stream = new Throttle(500e3);
  source.pipe(stream)


  // WITH THIS CODE IT WORKS AS EXPECTED
  /*
  fs.writeFileSync('./tmpfile', buf);
  const stream = fs.createReadStream('./tmpfile')
  */


  let hadFirstProgress = false;
  const tus = new Upload(stream, {
    endpoint: 'https://api2.transloadit.com/resumable/files/',
    uploadLengthDeferred: false,
    retryDelays: [0, 1000, 3000, 5000],
    uploadSize,
    chunkSize: 50e6,
    addRequestId: true,
    metadata: {
      filename: '10625646044fe84a0be97a4fd5c42181.mp4',
      filetype: 'video/mp4',
      username: 'John',
      license: 'Creative Commons',
      name: '10625646044fe84a0be97a4fd5c42181.mp4',
      type: 'video/mp4',
      check_test: '1',
      yo: '1',
      bla: '12333',
      assembly_url: assemblyUrl,
      fieldname: 'file'
    },
    onError (error) {
      console.log(error)
    },
    onProgress (bytesUploaded, bytesTotal) {
      if (!hadFirstProgress) onFirstProgress();
      hadFirstProgress = true;
      console.log({ bytesUploaded, bytesTotal })
    },
    onSuccess (data) {
      console.log('success', data)
    },
  })

  tus.start()

  function onFirstProgress() {
    setTimeout(() => {
      console.log('aborting')
      tus.abort();
    }, 2000)

    setTimeout(() => {
      console.log('resuming')
      tus.start()
    }, 5000)
  }
})().catch(console.error)

Now create an assembly in the state UPLOADING and replace assemblyUrl with the assembly’s url.

node index.js

Observe the log output:

{ bytesUploaded: 0, bytesTotal: 3000000 }
{ bytesUploaded: 16384, bytesTotal: 3000000 }
{ bytesUploaded: 50000, bytesTotal: 3000000 }
{ bytesUploaded: 180224, bytesTotal: 3000000 }
{ bytesUploaded: 250000, bytesTotal: 3000000 }
{ bytesUploaded: 409600, bytesTotal: 3000000 }
{ bytesUploaded: 458752, bytesTotal: 3000000 }
{ bytesUploaded: 507904, bytesTotal: 3000000 }
{ bytesUploaded: 606208, bytesTotal: 3000000 }
{ bytesUploaded: 655360, bytesTotal: 3000000 }
{ bytesUploaded: 753664, bytesTotal: 3000000 }
{ bytesUploaded: 819200, bytesTotal: 3000000 }
{ bytesUploaded: 900000, bytesTotal: 3000000 }
{ bytesUploaded: 950272, bytesTotal: 3000000 }
{ bytesUploaded: 1015808, bytesTotal: 3000000 }
{ bytesUploaded: 1064960, bytesTotal: 3000000 }
{ bytesUploaded: 1130496, bytesTotal: 3000000 }
{ bytesUploaded: 1163264, bytesTotal: 3000000 }
aborting
resuming
{ bytesUploaded: 1146810, bytesTotal: 3000000 }
{ bytesUploaded: 1261568, bytesTotal: 3000000 }
{ bytesUploaded: 1261568, bytesTotal: 3000000 }
{ bytesUploaded: 1277952, bytesTotal: 3000000 }
{ bytesUploaded: 1277952, bytesTotal: 3000000 }
{ bytesUploaded: 1283616, bytesTotal: 3000000 }
{ bytesUploaded: 1283616, bytesTotal: 3000000 }
{ bytesUploaded: 1283616, bytesTotal: 3000000 }
{ bytesUploaded: 1289280, bytesTotal: 3000000 }
{ bytesUploaded: 1289280, bytesTotal: 3000000 }
{ bytesUploaded: 1294336, bytesTotal: 3000000 }
{ bytesUploaded: 1294336, bytesTotal: 3000000 }
{ bytesUploaded: 1300000, bytesTotal: 3000000 }
{ bytesUploaded: 1300000, bytesTotal: 3000000 }
{ bytesUploaded: 1321440, bytesTotal: 3000000 }
{ bytesUploaded: 1321440, bytesTotal: 3000000 }
{ bytesUploaded: 1310720, bytesTotal: 3000000 }
{ bytesUploaded: 1310720, bytesTotal: 3000000 }
{ bytesUploaded: 1316384, bytesTotal: 3000000 }
{ bytesUploaded: 1337824, bytesTotal: 3000000 }
{ bytesUploaded: 1337824, bytesTotal: 3000000 }
{ bytesUploaded: 1386976, bytesTotal: 3000000 }
{ bytesUploaded: 1386976, bytesTotal: 3000000 }
{ bytesUploaded: 1332768, bytesTotal: 3000000 }
{ bytesUploaded: 1354208, bytesTotal: 3000000 }
{ bytesUploaded: 1403360, bytesTotal: 3000000 }
{ bytesUploaded: 1403360, bytesTotal: 3000000 }
{ bytesUploaded: 1468896, bytesTotal: 3000000 }
{ bytesUploaded: 1468896, bytesTotal: 3000000 }
{ bytesUploaded: 1360720, bytesTotal: 3000000 }
{ bytesUploaded: 1409872, bytesTotal: 3000000 }
{ bytesUploaded: 1475408, bytesTotal: 3000000 }
{ bytesUploaded: 1475408, bytesTotal: 3000000 }
{ bytesUploaded: 1507968, bytesTotal: 3000000 }
{ bytesUploaded: 1507968, bytesTotal: 3000000 }
{ bytesUploaded: 1419744, bytesTotal: 3000000 }
{ bytesUploaded: 1485280, bytesTotal: 3000000 }
{ bytesUploaded: 1517840, bytesTotal: 3000000 }
{ bytesUploaded: 1517840, bytesTotal: 3000000 }
{ bytesUploaded: 1577072, bytesTotal: 3000000 }
{ bytesUploaded: 1577072, bytesTotal: 3000000 }
{ bytesUploaded: 1436128, bytesTotal: 3000000 }
{ bytesUploaded: 1501664, bytesTotal: 3000000 }
{ bytesUploaded: 1534224, bytesTotal: 3000000 }
{ bytesUploaded: 1593456, bytesTotal: 3000000 }
{ bytesUploaded: 1593456, bytesTotal: 3000000 }
{ bytesUploaded: 1708144, bytesTotal: 3000000 }
{ bytesUploaded: 1708144, bytesTotal: 3000000 }
(node:65178) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 end listeners added to [Throttle]. Use emitter.setMaxListeners() to increase limit
{ bytesUploaded: 1550608, bytesTotal: 3000000 }
{ bytesUploaded: 1609840, bytesTotal: 3000000 }
{ bytesUploaded: 1724528, bytesTotal: 3000000 }
{ bytesUploaded: 1724528, bytesTotal: 3000000 }
{ bytesUploaded: 1822832, bytesTotal: 3000000 }
{ bytesUploaded: 1822832, bytesTotal: 3000000 }
(node:65178) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 data listeners added to [Throttle]. Use emitter.setMaxListeners() to increase limit
{ bytesUploaded: 1617200, bytesTotal: 3000000 }
{ bytesUploaded: 1731888, bytesTotal: 3000000 }
{ bytesUploaded: 1830192, bytesTotal: 3000000 }
...many more progres events...
{ bytesUploaded: 2920560, bytesTotal: 3000000 }
{ bytesUploaded: 2979984, bytesTotal: 3000000 }
{ bytesUploaded: 2979984, bytesTotal: 3000000 }
{ bytesUploaded: 3000000, bytesTotal: 3000000 }
success undefined

Note: the file does get uploaded successfully, it just takes a very long time, and it leaks EventEmitters

Alternatively, easier to reproduce but different error/outcome:

• Either replace endpoint: 'https://tusd.tusdemo.net/files/. No assemblyUrl needed. The upload completely hangs but crashes with a http 499 after a while
• Or replace endpoint: 'http://127.0.0.1:1080/files/ and enable/uncomment the tus-node-server code. This also causes the upload to slow to a halt (even slower progress events).

Even without Throttle, the same issue happens. I also tried with a fs.createReadStream pipe through a PassThrough stream, and same issue.

Expected behavior It should continue uploading with normal speed after calling start() again, just like for file streams.

Setup details Please provide following details, if applicable to your situation:

• Node 12
• Used tus-js-client version: latest
• Used tus server software: https://api2.transloadit.com/resumable/files/, https://tusd.tusdemo.net/files/, tus-node-server