@tus/s3-store: add termination extension (#401)

Author: naumf

README.md

@@ -110,7 +110,7 @@ The tus protocol supports optional [extensions][]. Below is a table of the suppo
 | [Creation With Upload][] | ✅                                | ✅                            | ✅                              |
 | [Expiration][]           | ✅                                | ❌                            | ❌                              |
 | [Checksum][]             | ❌                                | ❌                            | ❌                              |
-| [Termination][]          | ✅                                | ❌                            | ❌                              |
+| [Termination][]          | ✅                                | ✅                            | ❌                              |
 | [Concatenation][]        | ❌                                | ❌                            | ❌                              |
 
 ## Demos

packages/s3-store/README.md

@@ -67,7 +67,7 @@ but may increase it to not exceed the S3 10K parts limit.
 
 Options to pass to the AWS S3 SDK.
 Checkout the [`S3ClientConfig`](https://docs.aws.amazon.com/AWSJavaScriptSDK/v3/latest/clients/client-s3/interfaces/s3clientconfig.html)
-docs for the supported options. You need to at least set the `region`, `bucket` name, and your preferred method of authentication. 
+docs for the supported options. You need to at least set the `region`, `bucket` name, and your preferred method of authentication.
 
 ## Extensions
 
@@ -79,9 +79,13 @@ The tus protocol supports optional [extensions][]. Below is a table of the suppo
 | [Creation With Upload][] | ✅              |
 | [Expiration][]           | ❌              |
 | [Checksum][]             | ❌              |
-| [Termination][]          | ❌              |
+| [Termination][]          | ✅              |
 | [Concatenation][]        | ❌              |
 
+### Termination
+
+After a multipart upload is aborted, no additional parts can be uploaded using that upload ID. The storage consumed by any previously uploaded parts will be freed. However, if any part uploads are currently in progress, those part uploads might or might not succeed. As a result, it might be necessary to set an [S3 Lifecycle configuration](https://docs.aws.amazon.com/AmazonS3/latest/userguide/mpu-abort-incomplete-mpu-lifecycle-config.html) to abort incomplete multipart uploads.
+
 ## Examples
 
 ### Example: using `credentials` to fetch credentials inside a AWS container

packages/s3-store/index.ts

@@ -72,7 +72,12 @@ export class S3Store extends DataStore {
     super()
     const {partSize, s3ClientConfig} = options
     const {bucket, ...restS3ClientConfig} = s3ClientConfig
-    this.extensions = ['creation', 'creation-with-upload', 'creation-defer-length']
+    this.extensions = [
+      'creation',
+      'creation-with-upload',
+      'creation-defer-length',
+      'termination',
+    ]
     this.bucket = bucket
     this.preferredPartSize = partSize || 8 * 1024 * 1024
     this.client = new aws.S3(restS3ClientConfig)
@@ -517,4 +522,36 @@ export class S3Store extends DataStore {
 
     this.saveMetadata(file, upload_id)
   }
+
+  public async remove(id: string): Promise<void> {
+    try {
+      const {upload_id} = await this.getMetadata(id)
+      if (upload_id) {
+        await this.client
+          .abortMultipartUpload({
+            Bucket: this.bucket,
+            Key: id,
+            UploadId: upload_id,
+          })
+          .promise()
+      }
+    } catch (error) {
+      if (error?.code && ['NoSuchKey', 'NoSuchUpload'].includes(error.code)) {
+        log('remove: No file found.', error)
+        throw ERRORS.FILE_NOT_FOUND
+      }
+      throw error
+    }
+
+    await this.client
+      .deleteObjects({
+        Bucket: this.bucket,
+        Delete: {
+          Objects: [{Key: id}, {Key: `${id}.info`}],
+        },
+      })
+      .promise()
+
+    this.clearCache(id)
+  }
 }

packages/s3-store/test.ts

@@ -73,8 +73,7 @@ describe('S3DataStore', function () {
 
   shared.shouldHaveStoreMethods()
   shared.shouldCreateUploads()
-  // Termination extension not implemented yet
-  // shared.shouldRemoveUploads()
+  shared.shouldRemoveUploads() // Termination extension
   shared.shouldWriteUploads()
   shared.shouldHandleOffset()
   shared.shouldDeclareUploadLength() // Creation-defer-length extension

test/stores.test.ts

@@ -83,6 +83,37 @@ export const shouldRemoveUploads = function () {
       await this.datastore.create(file)
       return this.datastore.remove(file.id)
     })
+
+    it('should delete the file during upload', async function () {
+      const file = new Upload({
+        id: 'termination-test',
+        size: this.testFileSize,
+        offset: 0,
+        metadata: {filename: 'terminate_during_upload.pdf', is_confidential: null},
+      })
+      await this.datastore.create(file)
+
+      const readable = fs.createReadStream(this.testFilePath, {highWaterMark: 100 * 1024})
+      // Pause between chunks read to make sure that file is still uploading when terminate function is invoked
+      readable.on('data', () => {
+        readable.pause()
+        setTimeout(() => readable.resume(), 1000)
+      })
+
+      await Promise.allSettled([
+        this.datastore.write(readable, file.id, 0),
+        this.datastore.remove(file.id),
+      ])
+
+      try {
+        await this.datastore.getUpload(file.id)
+        assert.fail('getUpload should have thrown an error')
+      } catch (error) {
+        assert.equal([404, 410].includes(error?.status_code), true)
+      }
+
+      readable.destroy()
+    })
   })
 }
 
@@ -93,7 +124,7 @@ export const shouldWriteUploads = function () {
       return this.datastore.write(stream, 'doesnt_exist', 0).should.be.rejected()
     })
 
-    it('should reject whean readable stream has an error', async function () {
+    it('should reject when readable stream has an error', async function () {
       const stream = fs.createReadStream(this.testFilePath)
       return this.datastore.write(stream, 'doesnt_exist', 0).should.be.rejected()
     })