Datei-Uploads mit Web Workers und begrenzten Streams

mkdir worker-stream-demo && cd worker-stream-demo

<!doctype html>
<html lang="en">
  <head>
    <meta charset="utf-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <title>Worker stream uploads</title>
  </head>
  <body>
    <main>
      <h1>Worker stream uploads</h1>
      <label for="file-input">Up to four files, each at most 8 MiB</label>
      <input id="file-input" type="file" multiple />
      <button id="upload-files" type="button">Upload files</button>
      <button id="cancel-uploads" type="button" disabled>Cancel uploads</button>
      <p id="status" role="status">Choose files.</p>
      <ul id="results" aria-label="Upload results"></ul>
    </main>
    <script type="module" src="/main.js"></script>
  </body>
</html>

export type WorkerReply =
  | { type: 'progress'; progress: number; message: string }
  | { type: 'complete'; message: string; url: string }
  | { type: 'error'; message: string }

type Callbacks = {
  onProgress?: (progress: number, message: string) => void
  onComplete?: (message: string, url: string) => void
  onError?: (message: string) => void
}
type Task = { id: string; file: File }

class WorkerPool {
  closed = false
  workers: Worker[] = []
  idleWorkers: Worker[] = []
  taskQueue: Task[] = []
  taskCallbacks = new Map<string, Callbacks>()
  currentTasks = new Map<Worker, Task>()

  constructor(script: string, size = 2) {
    if (!Number.isInteger(size) || size < 1 || size > 2) throw new Error('Use one or two workers')
    try {
      for (let index = 0; index < size; index++) {
        const worker = new Worker(script, { type: 'module' })
        worker.onmessage = (event: MessageEvent<WorkerReply>) => this.handleWorkerMessage(worker, event.data)
        worker.onerror = (event) => {
          event.preventDefault()
          this.terminate('An upload worker failed. Try again.')
        }
        worker.onmessageerror = () => this.terminate('An upload worker failed. Try again.')
        this.workers.push(worker)
        this.idleWorkers.push(worker)
      }
    } catch (error) {
      this.terminate()
      throw error
    }
  }

  processFile(file: File, callbacks: Callbacks, id = crypto.randomUUID()): string {
    if (this.closed) throw new Error('The upload pool is closed.')
    if (this.taskCallbacks.size >= 4) throw new Error('The upload queue is full.')
    const task = { id, file }
    this.taskCallbacks.set(task.id, callbacks)
    const worker = this.idleWorkers.pop()
    if (worker) this.runTask(worker, task)
    else this.taskQueue.push(task)
    return task.id
  }

  runTask(worker: Worker, task: Task): void {
    this.currentTasks.set(worker, task)
    try {
      worker.postMessage(task)
    } catch {
      this.terminate('An upload worker failed. Try again.')
    }
  }

  handleWorkerMessage(worker: Worker, data: WorkerReply): void {
    if (this.closed) return
    const task = this.currentTasks.get(worker)
    if (!task) return
    const callbacks = this.taskCallbacks.get(task.id)
    if (!callbacks) return
    if (data.type === 'progress') {
      callbacks.onProgress?.(data.progress, data.message)
      return
    }
    this.currentTasks.delete(worker)
    this.taskCallbacks.delete(task.id)
    try {
      if (data.type === 'complete') callbacks.onComplete?.(data.message, data.url)
      else callbacks.onError?.(data.message)
    } finally {
      if (!this.closed) {
        const next = this.taskQueue.shift()
        if (next) this.runTask(worker, next)
        else this.idleWorkers.push(worker)
      }
    }
  }

  terminate(message = 'Upload canceled.'): void {
    if (this.closed) return
    this.closed = true
    for (const worker of this.workers) worker.terminate()
    this.workers = []
    this.idleWorkers = []
    this.taskQueue = []
    this.currentTasks.clear()
    const callbacks = [...this.taskCallbacks.values()]
    this.taskCallbacks.clear()
    const errors: unknown[] = []
    for (const callback of callbacks) {
      try {
        callback.onError?.(message)
      } catch (error) {
        errors.push(error)
      }
    }
    if (errors.length > 0) throw new AggregateError(errors, 'Upload cancellation callbacks failed.')
  }
}

const input = document.getElementById('file-input')
const upload = document.getElementById('upload-files')
const cancel = document.getElementById('cancel-uploads')
const status = document.getElementById('status')
const results = document.getElementById('results')
if (!(input instanceof HTMLInputElement) || !(upload instanceof HTMLButtonElement) ||
    !(cancel instanceof HTMLButtonElement) || !status || !results) throw new Error('Missing controls')

let pool: WorkerPool | null = null
let running = false
cancel.addEventListener('click', () => pool?.terminate())
window.addEventListener('pagehide', () => pool?.terminate())
input.addEventListener('change', () => {
  if (!running) status.textContent = 'Ready to upload.'
})

upload.addEventListener('click', async () => {
  if (running) return
  const files = Array.from(input.files ?? [])
  if (files.length < 1 || files.length > 4 || files.some((file) => file.size > 8 * 1024 * 1024)) {
    status.textContent = 'Choose one to four files, each at most 8 MiB.'
    return
  }
  running = true
  input.disabled = upload.disabled = true
  cancel.disabled = false
  results.replaceChildren()
  status.textContent = 'Uploading…'
  const failedIds: string[] = []
  let failures = 0
  try {
    const batchPool = new WorkerPool('/upload-worker.js')
    pool = batchPool
    const jobs = files.map((file) => new Promise<void>((resolve) => {
      const id = crypto.randomUUID()
      const row = document.createElement('li')
      const name = document.createElement('p')
      name.textContent = file.name
      const progress = document.createElement('progress')
      progress.max = 100
      progress.value = 0
      progress.setAttribute('aria-label', `Acknowledged bytes for ${file.name}`)
      const message = document.createElement('p')
      message.setAttribute('role', 'status')
      message.textContent = 'Queued.'
      row.append(name, progress, message)
      results.append(row)
      batchPool.processFile(file, {
        onProgress(value, text) { progress.value = value; message.textContent = text },
        onComplete(text, url) {
          progress.value = 100
          message.textContent = text
          const link = document.createElement('a')
          link.href = url
          link.download = 'upload.bin'
          link.textContent = `Download ${file.name}`
          row.append(link)
          resolve()
        },
        onError(text) {
          failures++
          failedIds.push(id)
          message.textContent = text
          resolve()
        },
      }, id)
    }))
    await Promise.all(jobs)
    status.textContent = failures === 0 ? 'All uploads complete.' : 'Some uploads did not complete.'
  } catch {
    status.textContent = 'Could not start the upload workers. Try again.'
  } finally {
    pool?.terminate()
    pool = null
    let cleanupFailed = false
    for (const id of failedIds) {
      try {
        const response = await fetch(`/uploads/${id}`, {
          method: 'DELETE', headers: { 'X-Upload-Demo': '1' }, signal: AbortSignal.timeout(5000),
        })
        if (!response.ok) cleanupFailed = true
      } catch { cleanupFailed = true }
    }
    if (cleanupFailed) status.textContent += ' Cleanup was not confirmed; uploads will expire.'
    running = false
    input.disabled = upload.disabled = false
    cancel.disabled = true
  }
})

import type { WorkerReply } from './main.js'

declare const self: DedicatedWorkerGlobalScope
const CHUNK_SIZE = 1024 * 1024

async function* readChunks(file: Blob): AsyncGenerator<Uint8Array<ArrayBuffer>> {
  const reader = file.stream().getReader()
  let buffer = new Uint8Array(CHUNK_SIZE)
  let used = 0
  let finished = false
  try {
    while (true) {
      const { done, value } = await reader.read()
      if (done) { finished = true; break }
      let offset = 0
      while (offset < value.length) {
        const length = Math.min(CHUNK_SIZE - used, value.length - offset)
        buffer.set(value.subarray(offset, offset + length), used)
        used += length
        offset += length
        if (used === CHUNK_SIZE) {
          yield buffer
          buffer = new Uint8Array(CHUNK_SIZE)
          used = 0
        }
      }
    }
    if (used > 0) yield buffer.subarray(0, used)
  } finally {
    if (!finished) await reader.cancel().catch(() => {})
    reader.releaseLock()
  }
}

function send(message: WorkerReply): void { self.postMessage(message) }

async function request(path: string, options: RequestInit): Promise<Response> {
  const response = await fetch(path, {
    ...options, headers: { ...options.headers, 'X-Upload-Demo': '1' },
    signal: AbortSignal.timeout(5000),
  })
  if (!response.ok) throw new Error('Upload request rejected')
  return response
}

self.onmessage = async (event: MessageEvent<{ file: File; id: string }>) => {
  const { file, id } = event.data
  try {
    if (!(file instanceof File) || file.size > 8 * CHUNK_SIZE) throw new Error('Invalid file')
    await request(`/uploads/${id}`, { method: 'POST', headers: { 'Upload-Length': String(file.size) } })
    send({ type: 'progress', progress: 0, message: 'Uploading…' })
    let acknowledged = 0
    for await (const chunk of readChunks(file)) {
      const form = new FormData()
      form.append('offset', String(acknowledged))
      form.append('chunk', new Blob([chunk]), 'chunk.bin')
      await request(`/uploads/${id}/chunks`, { method: 'POST', body: form })
      acknowledged += chunk.length
      send({ type: 'progress', progress: 100 * acknowledged / file.size,
        message: `${acknowledged} / ${file.size} bytes acknowledged.` })
    }
    send({ type: 'progress', progress: 100, message: 'Confirming…' })
    const response = await request(`/uploads/${id}/complete`, { method: 'POST' })
    if (await response.text() !== String(file.size)) throw new Error('Unexpected acknowledgment')
    send({ type: 'complete', message: 'Upload complete.', url: `/uploads/${id}/file` })
  } catch {
    send({ type: 'error', message: 'Upload failed. Try again.' })
  }
}

import { readFile } from 'node:fs/promises'
import { createServer } from 'node:http'
import type { IncomingMessage, ServerResponse } from 'node:http'
import { stripTypeScriptTypes } from 'node:module'

const CHUNK_SIZE = 1024 * 1024
const MAX_BODY = CHUNK_SIZE + 4096
const TTL_MS = 60_000
type Upload = { bytes: Buffer; received: number; complete: boolean; expires: number }
const uploads = new Map<string, Upload>()

class HttpError extends Error {
  status: number
  constructor(status: number, message: string) { super(message); this.status = status }
}

function expire(): void {
  for (const [id, upload] of uploads) {
    if (upload.expires <= Date.now()) uploads.delete(id)
  }
}

async function readBody(req: IncomingMessage, limit: number): Promise<Buffer> {
  const bytes = Buffer.alloc(limit)
  let length = 0
  // Keep the connection long enough to report an oversize body.
  for await (const part of req.iterator({ destroyOnReturn: false })) {
    if (!Buffer.isBuffer(part)) throw new HttpError(400, 'Expected bytes')
    if (length + part.length > limit) throw new HttpError(413, 'Body too large')
    part.copy(bytes, length)
    length += part.length
  }
  return bytes.subarray(0, length)
}

async function main(): Promise<void> {
  const assets = new Map<string, string>([['/', await readFile(new URL('./index.html', import.meta.url), 'utf8')]])
  for (const name of ['main', 'upload-worker']) {
    assets.set(`/${name}.js`, stripTypeScriptTypes(await readFile(new URL(`./${name}.ts`, import.meta.url), 'utf8')))
  }
  const port = Number(process.argv[2] ?? 0)
  if (!Number.isInteger(port) || port < 0 || port > 65535) throw new Error('Invalid port')
  let origin = ''
  let active = 0

  async function handle(req: IncomingMessage, res: ServerResponse): Promise<void> {
    if (req.headers.host !== new URL(origin).host ||
        (req.headers.origin !== undefined && req.headers.origin !== origin)) throw new HttpError(403, 'Use the printed local URL')
    const method = req.method
    if (method !== 'GET' && req.headers['x-upload-demo'] !== '1') throw new HttpError(403, 'Missing demo header')
    const path = new URL(req.url ?? '/', origin).pathname
    const isChunk = method === 'POST' && path.endsWith('/chunks')
    const body = await readBody(req, isChunk ? MAX_BODY : 0)
    // Parse before looking up an upload: pending work cannot retain a deleted session.
    let chunk: Uint8Array | undefined
    let offset: string | undefined
    if (isChunk) {
      const contentType = req.headers['content-type']
      if (typeof contentType !== 'string' || !contentType.startsWith('multipart/form-data;')) throw new HttpError(400, 'Expected multipart data')
      let form: FormData
      try {
        form = await new Request(origin, { method: 'POST', headers: { 'Content-Type': contentType }, body: new Uint8Array(body) }).formData()
      } catch { throw new HttpError(400, 'Invalid multipart data') }
      const value = form.get('chunk')
      const position = form.get('offset')
      if (!(value instanceof Blob) || value.size > CHUNK_SIZE || typeof position !== 'string' ||
          !/^(0|[1-9]\d*)$/.test(position) || [...form.keys()].length !== 2) throw new HttpError(400, 'Invalid chunk fields')
      chunk = new Uint8Array(await value.arrayBuffer())
      offset = position
    }
    expire()
    res.setHeader('Cache-Control', 'no-store')
    res.setHeader('X-Content-Type-Options', 'nosniff')
    const asset = assets.get(path)
    if (method === 'GET' && asset !== undefined) {
      res.setHeader('Content-Type', path === '/' ? 'text/html; charset=utf-8' : 'text/javascript')
      res.end(asset)
      return
    }
    const match = /^\/uploads\/([a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12})(?:\/(chunks|complete|file))?$/.exec(path)
    if (!match) throw new HttpError(404, 'Unknown route')
    const [, id, operation] = match
    if (method === 'POST' && operation === undefined) {
      const size = req.headers['upload-length']
      if (typeof size !== 'string' || !/^(0|[1-9]\d*)$/.test(size) || Number(size) > 8 * CHUNK_SIZE) throw new HttpError(400, 'Expected a size from 0 to 8 MiB')
      if (uploads.has(id)) throw new HttpError(409, 'Upload already exists')
      if (uploads.size >= 4) throw new HttpError(503, 'Upload capacity reached')
      uploads.set(id, { bytes: Buffer.alloc(Number(size)), received: 0, complete: false, expires: Date.now() + TTL_MS })
      res.writeHead(201).end()
      return
    }
    if (method === 'DELETE' && operation === undefined) {
      uploads.delete(id)
      res.writeHead(204).end()
      return
    }
    const upload = uploads.get(id)
    if (!upload) throw new HttpError(404, 'Upload missing or expired')
    if (isChunk && operation === 'chunks') {
      const expected = Math.min(CHUNK_SIZE, upload.bytes.length - upload.received)
      if (!chunk || chunk.length === 0 || chunk.length !== expected || offset !== String(upload.received) || upload.complete) throw new HttpError(409, 'Unexpected chunk offset or length')
      upload.bytes.set(chunk, upload.received)
      upload.received += chunk.length
      res.writeHead(204).end()
      return
    }
    if (method === 'POST' && operation === 'complete') {
      if (upload.received !== upload.bytes.length) throw new HttpError(409, 'Missing bytes')
      upload.complete = true
      res.end(String(upload.received))
      return
    }
    if (method === 'GET' && operation === 'file') {
      if (!upload.complete) throw new HttpError(409, 'Upload is not complete')
      res.setHeader('Content-Type', 'application/octet-stream')
      res.setHeader('Content-Disposition', 'attachment; filename="upload.bin"')
      res.end(upload.bytes)
      return
    }
    throw new HttpError(405, 'Unsupported operation')
  }

  const server = createServer({ requestTimeout: 10_000, headersTimeout: 10_000 }, (req, res) => {
    if (active >= 6) { res.writeHead(503, { Connection: 'close' }).end('Too many requests'); return }
    active++
    let handled = false
    let closed = false
    const deadline = setTimeout(() => { req.destroy(); res.destroy() }, 10_000)
    function release(): void {
      if (handled && closed) { clearTimeout(deadline); active-- }
    }
    res.once('close', () => { closed = true; release() })
    handle(req, res).catch((error: unknown) => {
      const status = error instanceof HttpError ? error.status : 500
      const message = error instanceof HttpError ? error.message : 'Request failed'
      if (!res.destroyed) res.writeHead(status, { Connection: 'close' }).end(message)
    }).finally(() => { handled = true; release() })
  })
  server.maxConnections = 16
  await new Promise<void>((resolve, reject) => {
    server.once('error', reject)
    server.listen(port, '127.0.0.1', () => resolve())
  })
  const address = server.address()
  if (!address || typeof address === 'string') throw new Error('Missing server address')
  origin = `http://127.0.0.1:${address.port}`
  setInterval(expire, 1000).unref()
  process.once('SIGINT', () => {
    server.close()
    server.closeAllConnections()
  })
  console.log(`Open ${origin}`)
}

main().catch((error: unknown) => {
  console.error(error instanceof Error ? error.message : 'Could not start the server')
  process.exitCode = 1
})

node server.mts