Watch upload stream + fix duplicate logs

.github/workflows/docs.yml

@@ -13,7 +13,7 @@ jobs:
       fail-fast: false
       matrix:
         platform: [ubuntu-22.04]
-        node: ['16']
+        node: ['20']
 
     runs-on: ${{ matrix.platform }}
 

src/http/error-handler.ts

@@ -43,6 +43,18 @@
         ? 500
         : 400
 
+      if (renderableError.code === ErrorCode.AbortedTerminate) {
+        reply.header('Connection', 'close')
+
+        reply.raw.once('finish', () => {
+          setTimeout(() => {
+            if (!request.raw.closed) {
+              request.raw.destroy()
+            }
+          }, 3000)
+        })
+      }
+
       return reply.status(statusCode).send({
         ...renderableError,
         error: error.error || renderableError.code,
@@ -52,14 +64,14 @@
     // Fastify errors
     if ('statusCode' in error) {
       const err = error as FastifyError
       return reply.status((error as any).statusCode || 500).send({
         statusCode: `${err.statusCode}`,
         error: err.name,
         message: err.message,
       })
     }
 
-    reply.status(500).send({
+    return reply.status(500).send({
       statusCode: '500',
       error: 'Internal',
       message: 'Internal Server Error',

src/http/plugins/log-request.ts

@@ -18,7 +18,7 @@
 
   interface FastifyContextConfig {
     operation?: { type: string }
     resources?: (req: FastifyRequest<any>) => string[]
   }
 }
 
@@ -32,10 +32,17 @@
       fastify.addHook('onRequest', async (req, res) => {
         req.startTime = Date.now()
 
-        // Request was aborted before the server finishes to return a response
         res.raw.once('close', () => {
-          const aborted = !res.raw.writableFinished
-          if (aborted) {
+          if (req.raw.aborted) {
+            doRequestLog(req, {
+              excludeUrls: options.excludeUrls,
+              statusCode: 'ABORTED REQ',
+              responseTime: (Date.now() - req.startTime) / 1000,
+            })
+            return
+          }
+
+          if (!res.raw.writableFinished) {
             doRequestLog(req, {
               excludeUrls: options.excludeUrls,
               statusCode: 'ABORTED RES',
@@ -53,7 +60,7 @@
         const resources = getFirstDefined<string[]>(
           req.resources,
           req.routeConfig.resources?.(req),
           (req.raw as any).resources,
           resourceFromParams ? [resourceFromParams] : ([] as string[])
         )
 
@@ -73,14 +80,6 @@
         }
       })
 
-      fastify.addHook('onRequestAbort', async (req) => {
-        doRequestLog(req, {
-          excludeUrls: options.excludeUrls,
-          statusCode: 'ABORTED REQ',
-          responseTime: (Date.now() - req.startTime) / 1000,
-        })
-      })
-
       fastify.addHook('onResponse', async (req, reply) => {
         doRequestLog(req, {
           reply,
@@ -116,7 +115,7 @@
   const rId = req.id
   const cIP = req.ip
   const statusCode = options.statusCode
   const error = (req.raw as any).executionError || req.executionError
   const tenantId = req.tenantId
 
   const buildLogMessage = `${tenantId} | ${rMeth} | ${statusCode} | ${cIP} | ${rId} | ${rUrl} | ${uAgent}`

src/http/plugins/signals.ts

@@ -20,6 +20,17 @@ export const signals = fastifyPlugin(
         disconnect: new AbortController(),
       }
 
+      // Client terminated the request before the body was fully sent
+      req.raw.once('close', () => {
+        if (req.raw.aborted) {
+          req.signals.body.abort()
+
+          if (!req.signals.disconnect.signal.aborted) {
+            req.signals.disconnect.abort()
+          }
+        }
+      })
+
       // Client terminated the request before server finished sending the response
       res.raw.once('close', () => {
         const aborted = !res.raw.writableFinished
@@ -33,7 +44,6 @@ export const signals = fastifyPlugin(
       })
     })
 
-    // Client terminated the request before the body was fully sent
     fastify.addHook('onRequestAbort', async (req) => {
       req.signals.body.abort()
 

src/http/plugins/tracing.ts

@@ -47,7 +47,7 @@ export const tracing = fastifyPlugin(
           if (
             tracingEnabled &&
             request.tracingMode &&
-            !['full', 'logs', 'debug'].includes(request.tracingMode)
+            !['logs', 'debug'].includes(request.tracingMode)
           ) {
             traceCollector.clearTrace(span.spanContext().traceId)
           }

src/internal/concurrency/index.ts

@@ -1,2 +1,3 @@
 export * from './mutex'
+export * from './stream'
 export * from './async-abort-controller'

src/internal/concurrency/stream.ts

@@ -0,0 +1,40 @@
+import { Transform, TransformCallback } from 'stream'
+
+interface ByteCounterStreamOptions {
+  maxHistory?: number
+  onMaxHistory?: (history: Date[]) => void
+  rewriteHistoryOnMax?: boolean
+}
+
+export const createByteCounterStream = (options: ByteCounterStreamOptions) => {
+  const { maxHistory = 100 } = options
+
+  let bytes = 0
+  let history: Date[] = []
+
+  const transformStream = new Transform({
+    transform(chunk: Buffer, encoding: string, callback: TransformCallback) {
+      bytes += chunk.length
+      history.push(new Date())
+
+      if (history.length === maxHistory) {
+        if (options.rewriteHistoryOnMax) {
+          options.onMaxHistory?.(history)
+          history = []
+        }
+      }
+
+      callback(null, chunk)
+    },
+  })
+
+  return {
+    transformStream,
+    get bytes() {
+      return bytes
+    },
+    get history() {
+      return history
+    },
+  }
+}

src/internal/errors/codes.ts

@@ -37,6 +37,7 @@ export enum ErrorCode {
   SlowDown = 'SlowDown',
   TusError = 'TusError',
   Aborted = 'Aborted',
+  AbortedTerminate = 'AbortedTerminate',
 }
 
 export const ERRORS = {
@@ -372,6 +373,13 @@ export const ERRORS = {
       message: message,
       originalError,
     }),
+  AbortedTerminate: (message: string, originalError?: unknown) =>
+    new StorageBackendError({
+      code: ErrorCode.AbortedTerminate,
+      httpStatusCode: 500,
+      message: message,
+      originalError,
+    }),
 }
 
 export function isStorageError(errorType: ErrorCode, error: any): error is StorageBackendError {

src/storage/backend/s3.ts

@@ -28,13 +28,16 @@ import { getSignedUrl } from '@aws-sdk/s3-request-presigner'
 import { ERRORS, StorageBackendError } from '@internal/errors'
 import { getConfig } from '../../config'
 import Agent, { HttpsAgent } from 'agentkeepalive'
-import { Readable } from 'stream'
+import { addAbortSignal, PassThrough, Readable } from 'node:stream'
 import {
   HttpPoolErrorGauge,
   HttpPoolFreeSocketsGauge,
   HttpPoolPendingRequestsGauge,
   HttpPoolSocketsGauge,
 } from '@internal/monitoring/metrics'
+import stream from 'stream/promises'
+import { trace } from '@opentelemetry/api'
+import { createByteCounterStream } from '@internal/concurrency'
 
 const { storageS3MaxSockets, region } = getConfig()
 
@@ -201,34 +204,83 @@ export class S3Backend implements StorageBackendAdapter {
     bucketName: string,
     key: string,
     version: string | undefined,
-    body: NodeJS.ReadableStream,
+    body: Readable,
     contentType: string,
     cacheControl: string,
     signal?: AbortSignal
   ): Promise<ObjectMetadata> {
+    if (signal?.aborted) {
+      throw ERRORS.Aborted('Upload was aborted')
+    }
+
+    const passThrough = new PassThrough()
+
+    if (signal) {
+      addAbortSignal(signal, passThrough)
+    }
+
+    passThrough.on('error', () => {
+      body.unpipe(passThrough)
+    })
+
+    body.on('error', (err) => {
+      if (!passThrough.closed) {
+        passThrough.destroy(err)
+      }
+    })
+
+    const currentAverage: number[] = []
+    const byteReader = createByteCounterStream({
+      maxHistory: 100,
+      rewriteHistoryOnMax: true,
+      onMaxHistory: (history) => {
+        currentAverage.push(averageTimeBetweenDates(history))
+      },
+    })
+    const bodyStream = body.pipe(passThrough)
+
+    let upload: Upload | undefined = undefined
+    let bytesUploaded = 0
+
     try {
-      const paralellUploadS3 = new Upload({
-        client: this.client,
-        params: {
-          Bucket: bucketName,
-          Key: withOptionalVersion(key, version),
-          /* @ts-expect-error: https://github.com/aws/aws-sdk-js-v3/issues/2085 */
-          Body: body,
-          ContentType: contentType,
-          CacheControl: cacheControl,
-        },
-      })
+      const data = await stream.pipeline(
+        bodyStream,
+        byteReader.transformStream,
+        async (bodyStream) => {
+          if (signal?.aborted) {
+            throw ERRORS.Aborted('Upload was aborted')
+          }
+
+          upload = new Upload({
+            client: this.client,
+            params: {
+              Bucket: bucketName,
+              Key: withOptionalVersion(key, version),
+              Body: bodyStream as Readable,
+              ContentType: contentType,
+              CacheControl: cacheControl,
+            },
+          })
+
+          upload.on('httpUploadProgress', (progress) => {
+            if (typeof progress.loaded !== 'undefined') {
+              bytesUploaded = progress.loaded
+            }
+          })
+
+          signal?.addEventListener(
+            'abort',
+            () => {
+              upload?.abort()
+            },
+            { once: true }
+          )
 
-      signal?.addEventListener(
-        'abort',
-        () => {
-          paralellUploadS3.abort()
+          return await upload.done()
         },
-        { once: true }
+        { signal }
       )
 
-      const data = await paralellUploadS3.done()
-
       const metadata = await this.headObject(bucketName, key, version)
 
       return {
@@ -242,6 +294,22 @@ export class S3Backend implements StorageBackendAdapter {
         contentRange: metadata.contentRange,
       }
     } catch (err: any) {
+      if (err instanceof Error && err.name === 'AbortError') {
+        const span = trace.getActiveSpan()
+        if (span) {
+          // Print how far we got uploading the file
+          span.setAttributes({
+            byteRead: byteReader.bytes,
+            bytesUploaded,
+            chunkTimes: JSON.stringify([
+              ...currentAverage,
+              averageTimeBetweenDates(byteReader.history),
+            ]),
+          })
+        }
+
+        throw ERRORS.AbortedTerminate('Upload was aborted', err)
+      }
       throw StorageBackendError.fromError(err)
     }
   }
@@ -409,22 +477,30 @@ export class S3Backend implements StorageBackendAdapter {
     length?: number,
     signal?: AbortSignal
   ) {
-    const paralellUploadS3 = new UploadPartCommand({
-      Bucket: bucketName,
-      Key: `${key}/${version}`,
-      UploadId: uploadId,
-      PartNumber: partNumber,
-      Body: body,
-      ContentLength: length,
-    })
+    try {
+      const paralellUploadS3 = new UploadPartCommand({
+        Bucket: bucketName,
+        Key: `${key}/${version}`,
+        UploadId: uploadId,
+        PartNumber: partNumber,
+        Body: body,
+        ContentLength: length,
+      })
 
-    const resp = await this.client.send(paralellUploadS3, {
-      abortSignal: signal,
-    })
+      const resp = await this.client.send(paralellUploadS3, {
+        abortSignal: signal,
+      })
 
-    return {
-      version,
-      ETag: resp.ETag,
+      return {
+        version,
+        ETag: resp.ETag,
+      }
+    } catch (e) {
+      if (e instanceof Error && e.name === 'AbortError') {
+        throw ERRORS.AbortedTerminate('Upload was aborted', e)
+      }
+
+      throw StorageBackendError.fromError(e)
     }
   }
 
@@ -531,3 +607,19 @@ export class S3Backend implements StorageBackendAdapter {
     return new S3Client(params)
   }
 }
+
+function averageTimeBetweenDates(dates: Date[]): number {
+  if (dates.length < 2) {
+    throw new Error('At least two dates are required to calculate the average time between them.')
+  }
+
+  let totalDifference = 0
+
+  for (let i = 1; i < dates.length; i++) {
+    const diff = dates[i].getTime() - dates[i - 1].getTime()
+    totalDifference += diff
+  }
+
+  const averageDifference = totalDifference / (dates.length - 1)
+  return averageDifference
+}