Harmonise error behaviour between the APIs that use BatchExecuteStatement. (#61)

Author: tachyonics

README.md

@@ -420,20 +420,6 @@ let constraintList: [TestPolymorphicTransactionConstraintEntry] = [
 try await table.polymorphicTransactWrite(entryList, constraints: constraintList)
 ```
 
-Both the `PolymorphicWriteEntry` and `PolymorphicTransactionConstraintEntry` conforming types can
-optionally provide a `compositePrimaryKey` property that will allow the API to return more information
-about failed transactions. This is enabled by default when using the `@PolymorphicWriteEntry` and 
-`@PolymorphicTransactionConstraintEntry` macros but can be disabled by setting the 
-`passCompositePrimaryKey` argument.
-
-```swift
-@PolymorphicWriteEntry(passCompositePrimaryKey: false)
-enum TestPolymorphicWriteEntry {
-    case testTypeA(TestTypeAWriteEntry)
-    case testTypeB(TestTypeBWriteEntry)
-}
-```
-
 ## Recording updates in a historical partition
 
 This package contains a number of convenience functions for storing versions of a row in a historical partition

Sources/DynamoDBTables/AWSDynamoDBCompositePrimaryKeyTable+deleteItems.swift

@@ -33,10 +33,10 @@ private let maximumUpdatesPerExecuteStatement = 25
 
 /// DynamoDBTable conformance updateItems function
 public extension AWSDynamoDBCompositePrimaryKeyTable {
-    private func deleteChunkedItems(_ keys: [CompositePrimaryKey<some Any>]) async throws {
+    private func deleteChunkedItems(_ keys: [CompositePrimaryKey<some Any>]) async throws -> [DynamoDBClientTypes.BatchStatementResponse] {
         // if there are no keys, there is nothing to update
         guard keys.count > 0 else {
-            return
+            return []
         }
 
         let statements = try keys.map { existingKey -> DynamoDBClientTypes.BatchStatementRequest in
@@ -49,13 +49,15 @@ public extension AWSDynamoDBCompositePrimaryKeyTable {
         let executeInput = BatchExecuteStatementInput(statements: statements)
 
         let response = try await self.dynamodb.batchExecuteStatement(input: executeInput)
-        try throwOnBatchExecuteStatementErrors(response: response)
+        return response.responses ?? []
     }
 
-    private func deleteChunkedItems(_ existingItems: [TypedTTLDatabaseItem<some Any, some Any, some Any>]) async throws {
+    private func deleteChunkedItems(_ existingItems: [TypedTTLDatabaseItem<some Any, some Any, some Any>]) async throws
+        -> [DynamoDBClientTypes.BatchStatementResponse]
+    {
         // if there are no items, there is nothing to update
         guard existingItems.count > 0 else {
-            return
+            return []
         }
 
         let statements = try existingItems.map { existingItem -> DynamoDBClientTypes.BatchStatementRequest in
@@ -68,24 +70,45 @@ public extension AWSDynamoDBCompositePrimaryKeyTable {
         let executeInput = BatchExecuteStatementInput(statements: statements)
 
         let response = try await self.dynamodb.batchExecuteStatement(input: executeInput)
-        try throwOnBatchExecuteStatementErrors(response: response)
+        return response.responses ?? []
     }
 
     func deleteItems(forKeys keys: [CompositePrimaryKey<some Any>]) async throws {
         // BatchExecuteStatement has a maximum of 25 statements
         // This function handles pagination internally.
         let chunkedKeys = keys.chunked(by: maximumUpdatesPerExecuteStatement)
-        try await chunkedKeys.concurrentForEach { chunk in
-            try await self.deleteChunkedItems(chunk)
+        let zippedResponses = try await chunkedKeys.concurrentFlatMap { chunk in
+            let responses = try await self.deleteChunkedItems(chunk)
+
+            return zip(responses, chunk)
+        }
+
+        let errors = zippedResponses.compactMap { response, key in
+            response.error?.asDynamoDBTableError(partitionKey: key.partitionKey, sortKey: key.sortKey, entryCount: keys.count)
+        }
+
+        if !errors.isEmpty {
+            throw DynamoDBTableError.batchFailures(errors: errors.removeDuplicates())
         }
     }
 
     func deleteItems(existingItems: [TypedTTLDatabaseItem<some Any, some Any, some Any>]) async throws {
         // BatchExecuteStatement has a maximum of 25 statements
         // This function handles pagination internally.
         let chunkedItems = existingItems.chunked(by: maximumUpdatesPerExecuteStatement)
-        try await chunkedItems.concurrentForEach { chunk in
-            try await self.deleteChunkedItems(chunk)
+        let zippedResponses = try await chunkedItems.concurrentFlatMap { chunk in
+            let responses = try await self.deleteChunkedItems(chunk)
+
+            return zip(responses, chunk)
+        }
+
+        let errors = zippedResponses.compactMap { response, item in
+            response.error?.asDynamoDBTableError(partitionKey: item.compositePrimaryKey.partitionKey,
+                                                 sortKey: item.compositePrimaryKey.sortKey, entryCount: existingItems.count)
+        }
+
+        if !errors.isEmpty {
+            throw DynamoDBTableError.batchFailures(errors: errors.removeDuplicates())
         }
     }
 }

Sources/DynamoDBTables/AWSDynamoDBCompositePrimaryKeyTable+execute.swift

@@ -74,8 +74,8 @@ public extension AWSDynamoDBCompositePrimaryKeyTable {
         // ExecuteStatement API has a maximum limit on the number of decomposed read operations per request.
         // Caller of this function needs to handle pagination on their side.
         guard partitionKeys.count <= maximumKeysPerExecuteStatement else {
-            throw DynamoDBTableError.validationError(
-                reason: "Execute API has a maximum limit of \(maximumKeysPerExecuteStatement) partition keys per request.")
+            throw DynamoDBTableError.validation(partitionKey: nil, sortKey: nil,
+                                                message: "Execute API has a maximum limit of \(maximumKeysPerExecuteStatement) partition keys per request.")
         }
 
         let statement = self.getStatement(partitionKeys: partitionKeys,
@@ -143,8 +143,8 @@ public extension AWSDynamoDBCompositePrimaryKeyTable {
         // ExecuteStatement API has a maximum limit on the number of decomposed read operations per request.
         // Caller of this function needs to handle pagination on their side.
         guard partitionKeys.count <= maximumKeysPerExecuteStatement else {
-            throw DynamoDBTableError.validationError(
-                reason: "Execute API has a maximum limit of \(maximumKeysPerExecuteStatement) partition keys per request.")
+            throw DynamoDBTableError.validation(partitionKey: nil, sortKey: nil,
+                                                message: "Execute API has a maximum limit of \(maximumKeysPerExecuteStatement) partition keys per request.")
         }
 
         let statement = self.getStatement(partitionKeys: partitionKeys,
@@ -256,3 +256,90 @@ public extension AWSDynamoDBCompositePrimaryKeyTable {
         }
     }
 }
+
+extension DynamoDBClientTypes.BatchStatementError {
+    func asDynamoDBTableError(partitionKey: String, sortKey: String, entryCount: Int) -> DynamoDBTableError? {
+        guard let code = self.code else {
+            return nil
+        }
+
+        // https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchStatementError.html
+        return switch code {
+        case .accessdenied:
+            DynamoDBTableError.accessDenied(message: self.message)
+        case .conditionalcheckfailed:
+            DynamoDBTableError.conditionalCheckFailed(partitionKey: partitionKey,
+                                                      sortKey: sortKey,
+                                                      message: self.message)
+        case .duplicateitem:
+            DynamoDBTableError.duplicateItem(partitionKey: partitionKey, sortKey: sortKey,
+                                             message: self.message)
+        case .internalservererror:
+            DynamoDBTableError.internalServerError(message: self.message)
+        case .itemcollectionsizelimitexceeded:
+            DynamoDBTableError.itemCollectionSizeLimitExceeded(attemptedSize: entryCount,
+                                                               maximumSize: AWSDynamoDBLimits.maximumUpdatesPerTransactionStatement)
+        case .provisionedthroughputexceeded:
+            DynamoDBTableError.provisionedThroughputExceeded(message: self.message)
+        case .requestlimitexceeded:
+            DynamoDBTableError.requestLimitExceeded(message: self.message)
+        case .resourcenotfound:
+            DynamoDBTableError.resourceNotFound(partitionKey: partitionKey, sortKey: sortKey,
+                                                message: self.message)
+        case .throttlingerror:
+            DynamoDBTableError.throttling(message: self.message)
+        case .transactionconflict:
+            DynamoDBTableError.transactionConflict(message: self.message)
+        case .validationerror:
+            DynamoDBTableError.validation(partitionKey: partitionKey, sortKey: sortKey,
+                                          message: self.message)
+        case let .sdkUnknown(message):
+            DynamoDBTableError.unknown(code: message, partitionKey: partitionKey,
+                                       sortKey: sortKey, message: self.message)
+        }
+    }
+}
+
+extension [DynamoDBTableError] {
+    func removeDuplicates() -> [DynamoDBTableError] {
+        var seenAccessDenied = false
+        var seenInternalServerError = false
+        var seenRequestLimitExceeded = false
+        var seenStatementLengthExceeded = false
+        var seenItemCollectionSizeLimitExceeded = false
+        var seenProvisionedThroughputExceeded = false
+
+        func canPassThrough(state: inout Bool) -> Bool {
+            if state {
+                return false
+            } else {
+                state = true
+                return true
+            }
+        }
+
+        // iterate through all errors
+        return self.compactMap { error in
+            return switch error {
+            case .accessDenied:
+                canPassThrough(state: &seenAccessDenied) ? error : nil
+            case .internalServerError:
+                canPassThrough(state: &seenInternalServerError) ? error : nil
+            case .requestLimitExceeded:
+                canPassThrough(state: &seenRequestLimitExceeded) ? error : nil
+            case .statementLengthExceeded:
+                canPassThrough(state: &seenStatementLengthExceeded) ? error : nil
+            case .itemCollectionSizeLimitExceeded:
+                canPassThrough(state: &seenItemCollectionSizeLimitExceeded) ? error : nil
+            case .provisionedThroughputExceeded:
+                canPassThrough(state: &seenProvisionedThroughputExceeded) ? error : nil
+            case .conditionalCheckFailed, .duplicateItem, .concurrencyError, .validation, .throttling, .databaseError,
+                 .unexpectedError, .unexpectedResponse, .resourceNotFound, .typeMismatch, .batchAPIExceededRetries,
+                 .unexpectedType, .unableToUpdateError, .unrecognizedError, .multipleUnexpectedErrors, .transactionCanceled,
+                 .transactionConflict, .batchFailures, .unknown:
+                // always pass through these errors
+                error
+            }
+        }
+    }
+}