fix: make empty bucket use queue to remove underlying objects asynchronously

Author: itslenny

src/storage/events/index.ts

@@ -4,6 +4,7 @@ export * from './object-created'
 export * from './object-updated'
 export * from './object-removed'
 export * from './object-admin-delete'
+export * from './object-admin-delete-batch'
 export * from './backup-object'
 export * from './run-migrations'
 export * from './reset-migrations'

src/storage/events/object-admin-delete-batch.ts

@@ -0,0 +1,90 @@
+import { BaseEvent } from './base-event'
+import { getConfig } from '../../config'
+import { Job, SendOptions, WorkOptions } from 'pg-boss'
+import { logger, logSchema } from '@internal/monitoring'
+import { Storage } from '../index'
+import { BasePayload } from '@internal/queue'
+
+export interface ObjectDeleteBatchEvent extends BasePayload {
+  prefixes: string[]
+  bucketId: string
+}
+
+const { storageS3Bucket, adminDeleteQueueTeamSize, adminDeleteConcurrency } = getConfig()
+
+export class ObjectAdminDeleteBatch extends BaseEvent<ObjectDeleteBatchEvent> {
+  static queueName = 'object:admin:delete-batch'
+
+  static getWorkerOptions(): WorkOptions {
+    return {
+      teamSize: adminDeleteQueueTeamSize,
+      teamConcurrency: adminDeleteConcurrency,
+    }
+  }
+
+  static getQueueOptions(): SendOptions {
+    return {
+      priority: 10,
+    }
+  }
+
+  static async handle(job: Job<ObjectDeleteBatchEvent>) {
+    console.log('in ObjectAdminDeleteBatch.handle()')
+    let storage: Storage | undefined = undefined
+
+    const { prefixes, bucketId } = job.data
+    if (prefixes.length < 1) {
+      return
+    }
+
+    try {
+      storage = await this.createStorage(job.data)
+
+      logSchema.event(logger, `[Admin]: ObjectAdminDeleteBatch ${bucketId} ${prefixes.length}`, {
+        jodId: job.id,
+        type: 'event',
+        event: 'ObjectAdminDeleteBatch',
+        payload: JSON.stringify(job.data),
+        objectPath: bucketId,
+        resources: prefixes,
+        tenantId: job.data.tenant.ref,
+        project: job.data.tenant.ref,
+        reqId: job.data.reqId,
+      })
+
+      await storage.backend.deleteObjects(storageS3Bucket, prefixes)
+    } catch (e) {
+      logger.error(
+        {
+          error: e,
+          jodId: job.id,
+          type: 'event',
+          event: 'ObjectAdminDeleteBatch',
+          payload: JSON.stringify(job.data),
+          objectPath: bucketId,
+          resources: prefixes,
+          tenantId: job.data.tenant.ref,
+          project: job.data.tenant.ref,
+          reqId: job.data.reqId,
+        },
+        `[Admin]: ObjectAdminDeleteBatch ${bucketId} ${prefixes.length} - FAILED`
+      )
+      throw e
+    } finally {
+      if (storage) {
+        const tenant = storage.db.tenant()
+        storage.db
+          .destroyConnection()
+          .then(() => {
+            // no-op
+          })
+          .catch((e) => {
+            logger.error(
+              { error: e },
+              `[Admin]: ObjectAdminDeleteBatch ${tenant.ref} - FAILED DISPOSING CONNECTION`
+            )
+          })
+      }
+    }
+  }
+}

src/storage/events/workers.ts

@@ -1,6 +1,7 @@
 import { Queue } from '@internal/queue'
 import {
   ObjectAdminDelete,
+  ObjectAdminDeleteBatch,
   Webhook,
   RunMigrationsOnTenants,
   BackupObjectEvent,
@@ -11,6 +12,7 @@ import {
 export function registerWorkers() {
   Queue.register(Webhook)
   Queue.register(ObjectAdminDelete)
+  Queue.register(ObjectAdminDeleteBatch)
   Queue.register(RunMigrationsOnTenants)
   Queue.register(BackupObjectEvent)
   Queue.register(ResetMigrationsOnTenant)

src/storage/storage.ts

@@ -6,9 +6,9 @@ import { getFileSizeLimit, mustBeValidBucketName, parseFileSizeToBytes } from '.
 import { getConfig } from '../config'
 import { ObjectStorage } from './object'
 import { InfoRenderer } from '@storage/renderer/info'
-import { logger, logSchema } from '@internal/monitoring'
+import { ObjectAdminDeleteBatch } from './events'
 
-const { requestUrlLengthLimit, storageS3Bucket } = getConfig()
+const { requestUrlLengthLimit } = getConfig()
 
 /**
  * Storage
@@ -200,15 +200,18 @@ export class Storage {
       )
 
       if (deleted && deleted.length > 0) {
-        const params = deleted.reduce((all, { name, version }) => {
+        const prefixes = deleted.reduce((all, { name, version }) => {
           const fileName = withOptionalVersion(`${this.db.tenantId}/${bucketId}/${name}`, version)
           all.push(fileName)
           all.push(fileName + '.info')
           return all
         }, [] as string[])
         // delete files from s3 asynchronously
-        this.backend.deleteObjects(storageS3Bucket, params).catch((e) => {
-          logSchema.error(logger, 'Failed to delete objects from s3', { type: 's3', error: e })
+        await ObjectAdminDeleteBatch.send({
+          prefixes,
+          bucketId,
+          tenant: this.db.tenant(),
+          reqId: this.db.reqId,
         })
       }
 

src/test/bucket.test.ts

@@ -423,6 +423,23 @@ describe('testing EMPTY bucket', () => {
 
   test('user is able to empty a bucket with a service key', async () => {
     const bucketId = 'bucket3'
+
+    // confirm there are items in the bucket before empty
+    const responseList = await appInstance.inject({
+      method: 'POST',
+      url: '/object/list/' + bucketId,
+      headers: {
+        authorization: `Bearer ${process.env.SERVICE_KEY}`,
+      },
+      payload: {
+        prefix: '',
+        limit: 10,
+        offset: 0,
+      },
+    })
+    expect(responseList.statusCode).toBe(200)
+    expect(responseList.json()).toHaveLength(2)
+
     const response = await appInstance.inject({
       method: 'POST',
       url: `/bucket/${bucketId}/empty`,
@@ -433,6 +450,20 @@ describe('testing EMPTY bucket', () => {
     expect(response.statusCode).toBe(200)
     const responseJSON = JSON.parse(response.body)
     expect(responseJSON.message).toBe('Successfully emptied')
+
+    // confirm the bucket is actually empty after
+    const responseList2 = await appInstance.inject({
+      method: 'POST',
+      url: '/object/list/' + bucketId,
+      headers: {
+        authorization: `Bearer ${process.env.SERVICE_KEY}`,
+      },
+      payload: {
+        prefix: '',
+      },
+    })
+    expect(responseList2.statusCode).toBe(200)
+    expect(responseList2.json()).toHaveLength(0)
   })
 
   test('user is able to delete a bucket', async () => {