fix: gracefully handle S3 NoSuchUpload error when removing multipart upload, improve S3 error logging (#1199)

Author: itslenny

src/internal/errors/codes.test.ts

@@ -127,6 +127,23 @@ describe('normalizeRawError', () => {
     expect(result.raw).not.toContain('secret root cert')
     expect(JSON.parse(result.raw)).toEqual({ code: '08006' })
   })
+
+  it('handles S3 errors with correct errorCode and statusCode', () => {
+    const s3Error = new Error('The specified upload does not exist.')
+    s3Error.name = 'NoSuchUpload'
+    Object.assign(s3Error, {
+      $metadata: {
+        httpStatusCode: 404,
+      },
+    })
+
+    const result = normalizeRawError(s3Error, 'info')
+
+    expect(result.errorCode).toBe(ErrorCode.S3Error)
+    expect(result.statusCode).toBe(404)
+    expect(result.name).toBe('NoSuchUpload')
+    expect(result.message).toBe('The specified upload does not exist.')
+  })
 })
 
 describe('getErrorCode', () => {

src/internal/errors/codes.ts

@@ -1,7 +1,7 @@
 import { IcebergError } from '@storage/protocols/iceberg/catalog/errors'
 import { DatabaseError } from 'pg'
 import { configure } from 'safe-stable-stringify'
-import { StorageBackendError } from './storage-error'
+import { isS3Error, StorageBackendError } from './storage-error'
 
 export enum ErrorCode {
   NoSuchBucket = 'NoSuchBucket',
@@ -592,6 +592,9 @@ function hasStringErrorCode(error: Error): error is Error & { code: string } {
 }
 
 export function getErrorCode(error: Error): string {
+  if (isS3Error(error)) {
+    return ErrorCode.S3Error
+  }
   if (error instanceof IcebergError && error.error) {
     return error.error
   }
@@ -607,6 +610,9 @@ export function getErrorCode(error: Error): string {
 }
 
 function getErrorStatusCode(error: Error): number {
+  if (isS3Error(error) && error.$metadata.httpStatusCode) {
+    return error.$metadata.httpStatusCode
+  }
   if (error instanceof StorageBackendError && error.httpStatusCode) {
     return error.httpStatusCode
   }

src/storage/backend/s3/adapter.test.ts

@@ -1,4 +1,5 @@
 import {
+  AbortMultipartUploadCommand,
   DeleteObjectsCommand,
   GetObjectCommand,
   HeadObjectCommand,
@@ -547,4 +548,39 @@ describe('S3Backend', () => {
       }
     })
   })
+
+  describe('abortMultipartUpload', () => {
+    test('includes version in S3 key when version is provided', async () => {
+      const backend = createBackend()
+      const bucketName = 'test-bucket'
+      const key = 'test-folder/test-object.txt'
+      const uploadId = 'test-upload-id'
+      const version = 'version-123'
+
+      await backend.abortMultipartUpload(bucketName, key, uploadId, version)
+
+      expect(mockSend).toHaveBeenCalledTimes(1)
+      const command = mockSend.mock.calls[0][0] as AbortMultipartUploadCommand
+      expect(command).toBeInstanceOf(AbortMultipartUploadCommand)
+      expect(command.input.Bucket).toBe(bucketName)
+      expect(command.input.Key).toBe(`${key}/${version}`)
+      expect(command.input.UploadId).toBe(uploadId)
+    })
+
+    test('does not include version in S3 key when version is undefined', async () => {
+      const backend = createBackend()
+      const bucketName = 'test-bucket'
+      const key = 'test-folder/test-object.txt'
+      const uploadId = 'test-upload-id'
+
+      await backend.abortMultipartUpload(bucketName, key, uploadId, undefined)
+
+      expect(mockSend).toHaveBeenCalledTimes(1)
+      const command = mockSend.mock.calls[0][0] as AbortMultipartUploadCommand
+      expect(command).toBeInstanceOf(AbortMultipartUploadCommand)
+      expect(command.input.Bucket).toBe(bucketName)
+      expect(command.input.Key).toBe(key)
+      expect(command.input.UploadId).toBe(uploadId)
+    })
+  })
 })

src/storage/backend/s3/adapter.ts

@@ -580,7 +580,7 @@ export class S3Backend implements StorageBackendAdapter {
     try {
       const paralellUploadS3 = new UploadPartCommand({
         Bucket: bucketName,
-        Key: version ? `${key}/${version}` : key,
+        Key: withOptionalVersion(key, version),
         UploadId: uploadId,
         PartNumber: partNumber,
         Body: body,
@@ -617,7 +617,7 @@ export class S3Backend implements StorageBackendAdapter {
     if (parts.length === 0) {
       const listPartsInput = new ListPartsCommand({
         Bucket: bucketName,
-        Key: version ? key + '/' + version : key,
+        Key: withOptionalVersion(key, version),
         UploadId: uploadId,
       })
 
@@ -627,7 +627,7 @@ export class S3Backend implements StorageBackendAdapter {
 
     const completeUpload = new CompleteMultipartUploadCommand({
       Bucket: bucketName,
-      Key: version ? key + '/' + version : key,
+      Key: withOptionalVersion(key, version),
       UploadId: uploadId,
       MultipartUpload:
         parts.length === 0
@@ -658,10 +658,15 @@ export class S3Backend implements StorageBackendAdapter {
     }
   }
 
-  async abortMultipartUpload(bucketName: string, key: string, uploadId: string): Promise<void> {
+  async abortMultipartUpload(
+    bucketName: string,
+    key: string,
+    uploadId: string,
+    version?: string
+  ): Promise<void> {
     const abortUpload = new AbortMultipartUploadCommand({
       Bucket: bucketName,
-      Key: key,
+      Key: withOptionalVersion(key, version),
       UploadId: uploadId,
     })
     await this.client.send(abortUpload)

src/storage/protocols/s3/s3-handler.test.ts

@@ -156,3 +156,146 @@ describe('S3ProtocolHandler.getObject', () => {
     )
   })
 })
+
+describe('S3ProtocolHandler.abortMultipartUpload', () => {
+  it('aborts multipart upload and deletes from database when backend succeeds', async () => {
+    const uploadId = 'test-upload-id'
+    const findMultipartUpload = vi.fn().mockResolvedValue({
+      id: uploadId,
+      version: 'test-version',
+      user_metadata: { key: 'value' },
+      metadata: { mimetype: 'text/plain' },
+    })
+    const deleteMultipartUpload = vi.fn().mockResolvedValue(undefined)
+    const testPermission = vi.fn().mockResolvedValue(undefined)
+    const abortMultipartUpload = vi.fn().mockResolvedValue(undefined)
+    const getKeyLocation = vi.fn(() => 'tenant-id/bucket/object.txt')
+
+    const storage = {
+      backend: {
+        abortMultipartUpload,
+      },
+      db: {
+        asSuperUser: vi.fn(() => ({
+          findMultipartUpload,
+          deleteMultipartUpload,
+        })),
+        testPermission,
+      },
+      location: {
+        getKeyLocation,
+      },
+    }
+    const handler = new S3ProtocolHandler(storage as never, 'tenant-id')
+
+    const response = await handler.abortMultipartUpload({
+      Bucket: 'bucket',
+      Key: 'object.txt',
+      UploadId: uploadId,
+    })
+
+    expect(response).toEqual({})
+    expect(findMultipartUpload).toHaveBeenCalledWith(uploadId, 'id,version,user_metadata,metadata')
+    expect(abortMultipartUpload).toHaveBeenCalled()
+    expect(deleteMultipartUpload).toHaveBeenCalledWith(uploadId)
+  })
+
+  it('deletes from database when backend throws NoSuchUpload error', async () => {
+    const uploadId = 'test-upload-id'
+    const findMultipartUpload = vi.fn().mockResolvedValue({
+      id: uploadId,
+      version: 'test-version',
+      user_metadata: null,
+      metadata: null,
+    })
+    const deleteMultipartUpload = vi.fn().mockResolvedValue(undefined)
+    const testPermission = vi.fn().mockResolvedValue(undefined)
+    const noSuchUploadError = {
+      name: 'NoSuchUpload',
+      message: 'The specified upload does not exist.',
+      $metadata: {
+        httpStatusCode: 404,
+      },
+    }
+    const abortMultipartUpload = vi.fn().mockRejectedValue(noSuchUploadError)
+    const getKeyLocation = vi.fn(() => 'tenant-id/bucket/object.txt')
+
+    const storage = {
+      backend: {
+        abortMultipartUpload,
+      },
+      db: {
+        asSuperUser: vi.fn(() => ({
+          findMultipartUpload,
+          deleteMultipartUpload,
+        })),
+        testPermission,
+      },
+      location: {
+        getKeyLocation,
+      },
+    }
+    const handler = new S3ProtocolHandler(storage as never, 'tenant-id')
+
+    const response = await handler.abortMultipartUpload({
+      Bucket: 'bucket',
+      Key: 'object.txt',
+      UploadId: uploadId,
+    })
+
+    expect(response).toEqual({})
+    expect(findMultipartUpload).toHaveBeenCalledWith(uploadId, 'id,version,user_metadata,metadata')
+    expect(abortMultipartUpload).toHaveBeenCalled()
+    expect(deleteMultipartUpload).toHaveBeenCalledWith(uploadId)
+  })
+
+  it('throws error when backend throws non-NoSuchUpload error', async () => {
+    const uploadId = 'test-upload-id'
+    const findMultipartUpload = vi.fn().mockResolvedValue({
+      id: uploadId,
+      version: 'test-version',
+      user_metadata: null,
+      metadata: null,
+    })
+    const deleteMultipartUpload = vi.fn().mockResolvedValue(undefined)
+    const testPermission = vi.fn().mockResolvedValue(undefined)
+    const otherError = {
+      name: 'AccessDenied',
+      message: 'Access Denied',
+      $metadata: {
+        httpStatusCode: 403,
+      },
+    }
+    const abortMultipartUpload = vi.fn().mockRejectedValue(otherError)
+    const getKeyLocation = vi.fn(() => 'tenant-id/bucket/object.txt')
+
+    const storage = {
+      backend: {
+        abortMultipartUpload,
+      },
+      db: {
+        asSuperUser: vi.fn(() => ({
+          findMultipartUpload,
+          deleteMultipartUpload,
+        })),
+        testPermission,
+      },
+      location: {
+        getKeyLocation,
+      },
+    }
+    const handler = new S3ProtocolHandler(storage as never, 'tenant-id')
+
+    await expect(
+      handler.abortMultipartUpload({
+        Bucket: 'bucket',
+        Key: 'object.txt',
+        UploadId: uploadId,
+      })
+    ).rejects.toEqual(otherError)
+
+    expect(findMultipartUpload).toHaveBeenCalledWith(uploadId, 'id,version,user_metadata,metadata')
+    expect(abortMultipartUpload).toHaveBeenCalled()
+    expect(deleteMultipartUpload).not.toHaveBeenCalled()
+  })
+})

src/storage/protocols/s3/s3-handler.ts

@@ -18,7 +18,7 @@ import {
   UploadPartCopyCommandInput,
 } from '@aws-sdk/client-s3'
 import { decrypt, encrypt } from '@internal/auth'
-import { ERRORS, ErrorCode, isStorageError } from '@internal/errors'
+import { ERRORS, ErrorCode, isS3Error, isStorageError } from '@internal/errors'
 import { isValidHeader } from '@internal/http/header'
 import { logger, logSchema } from '@internal/monitoring'
 import { PassThrough, Readable } from 'stream'
@@ -787,16 +787,25 @@ export class S3ProtocolHandler {
       metadata: multipart.metadata || undefined,
     })
 
-    await this.storage.backend.abortMultipartUpload(
-      storageS3Bucket,
-      this.storage.location.getKeyLocation({
-        bucketId: Bucket,
-        objectName: Key,
-        tenantId: this.tenantId,
-      }),
-      UploadId,
-      multipart.version
-    )
+    try {
+      await this.storage.backend.abortMultipartUpload(
+        storageS3Bucket,
+        this.storage.location.getKeyLocation({
+          bucketId: Bucket,
+          objectName: Key,
+          tenantId: this.tenantId,
+        }),
+        UploadId,
+        multipart.version
+      )
+    } catch (e) {
+      // gracefully continue if the upload part was already deleted/aborted on the S3 side
+      // error.name: NoSuchUpload
+      // error.message: The specified upload does not exist. The upload ID may be invalid, or the upload may have been aborted or completed.
+      if (!isS3Error(e) || e.name !== 'NoSuchUpload') {
+        throw e
+      }
+    }
 
     await this.storage.db.asSuperUser().deleteMultipartUpload(UploadId)