From dda72abddb95664a9a0403cb32a10367a3b82e42 Mon Sep 17 00:00:00 2001
From: Alejandro Corredor <alejandro.corredor@lifeomic.com>
Date: Thu, 3 Oct 2024 17:02:35 -0400
Subject: [PATCH] fix: EOCI-213 [pt. 4] better error logging

---
 src/dynamo-streams.test.ts |  4 ++--
 src/kinesis.test.ts        |  4 ++--
 src/sqs.test.ts            |  8 ++++----
 src/utils.ts               | 25 ++++++++++++++-----------
 4 files changed, 22 insertions(+), 19 deletions(-)

diff --git a/src/dynamo-streams.test.ts b/src/dynamo-streams.test.ts
index 264facb..1a9de42 100644
--- a/src/dynamo-streams.test.ts
+++ b/src/dynamo-streams.test.ts
@@ -905,7 +905,7 @@ describe('DynamoStreamHandler', () => {
       // Expect that redaction is working
       expect(logger.error).toHaveBeenCalledWith(
         expect.objectContaining({
-          identifier: 'one',
+          itemIdentifier: 'one',
           failedRecord: {
             eventName: 'INSERT',
             dynamodb: {
@@ -926,7 +926,7 @@ describe('DynamoStreamHandler', () => {
       // expect that third event is logged
       expect(logger.error).toHaveBeenCalledWith(
         expect.objectContaining({
-          identifier: 'three',
+          itemIdentifier: 'three',
           failedRecord: {
             eventName: 'INSERT',
             dynamodb: {
diff --git a/src/kinesis.test.ts b/src/kinesis.test.ts
index fb71435..bb93141 100644
--- a/src/kinesis.test.ts
+++ b/src/kinesis.test.ts
@@ -539,7 +539,7 @@ describe('KinesisEventHandler', () => {
       // Expect that first event is logged
       expect(logger.error).toHaveBeenCalledWith(
         expect.objectContaining({
-          identifier: 'one',
+          itemIdentifier: 'one',
           failedRecord: {
             kinesis: {
               sequenceNumber: 'one',
@@ -559,7 +559,7 @@ describe('KinesisEventHandler', () => {
       // Expect that third event is logged
       expect(logger.error).toHaveBeenCalledWith(
         expect.objectContaining({
-          identifier: 'three',
+          itemIdentifier: 'three',
           failedRecord: {
             kinesis: {
               sequenceNumber: 'three',
diff --git a/src/sqs.test.ts b/src/sqs.test.ts
index 3a38476..5d369a0 100644
--- a/src/sqs.test.ts
+++ b/src/sqs.test.ts
@@ -371,7 +371,7 @@ describe('SQSMessageHandler', () => {
       // First failure group, expecting message bodies
       expect(logger.error).toHaveBeenCalledWith(
         expect.objectContaining({
-          identifier: 'message-3',
+          itemIdentifier: 'message-3',
           failedRecord: expect.objectContaining({
             body: JSON.stringify({
               name: `test-event-3`,
@@ -387,7 +387,7 @@ describe('SQSMessageHandler', () => {
       // Second failure group, expecting message bodies
       expect(logger.error).toHaveBeenCalledWith(
         expect.objectContaining({
-          identifier: 'message-7',
+          itemIdentifier: 'message-7',
           failedRecord: expect.objectContaining({
             body: JSON.stringify({
               name: `test-event-7`,
@@ -487,7 +487,7 @@ describe('SQSMessageHandler', () => {
       // First failure group, expecting message bodies are redacted
       expect(logger.error).toHaveBeenCalledWith(
         expect.objectContaining({
-          identifier: 'message-3',
+          itemIdentifier: 'message-3',
           failedRecord: expect.objectContaining({
             body: 'REDACTED',
             messageId: 'message-3',
@@ -502,7 +502,7 @@ describe('SQSMessageHandler', () => {
       // Second failure group, expecting message bodies are redacted
       expect(logger.error).toHaveBeenCalledWith(
         expect.objectContaining({
-          identifier: 'message-7',
+          itemIdentifier: 'message-7',
           failedRecord: expect.objectContaining({
             body: 'REDACTED',
             messageId: 'message-7',
diff --git a/src/utils.ts b/src/utils.ts
index b5b9cb6..94c8abc 100644
--- a/src/utils.ts
+++ b/src/utils.ts
@@ -163,29 +163,32 @@ export const handleUnprocessedRecords = <Record>(params: {
     return;
   }
 
-  if (!params.usePartialBatchResponses) {
-    throw new AggregateError(
-      params.unprocessedRecords.filter((i) => 'error' in i).map((e) => e.error),
-    );
-  }
-
-  // Log all the failures.
+  const batchItemFailures: { itemIdentifier: string }[] = [];
+  const errors: any[] = [];
+  // Even when not using partial batch responses, we still want to log all
+  // the errors before throwing so we can easily correlate errors by the
+  // logger correlation ID, request ID, and event ID.
   for (const { item, error } of params.unprocessedRecords) {
+    const itemIdentifier = params.getItemIdentifier(item);
+    batchItemFailures.push({ itemIdentifier });
+
     if (error) {
+      errors.push(error);
       params.logger.error(
         {
           err: error,
-          identifier: params.getItemIdentifier(item),
+          itemIdentifier,
           failedRecord: params.redactRecord ? params.redactRecord(item) : item,
+          usePartialBatchResponses: params.usePartialBatchResponses,
         },
         'Failed to process record',
       );
     }
   }
 
-  const batchItemFailures = params.unprocessedRecords.map(({ item }) => ({
-    itemIdentifier: params.getItemIdentifier(item),
-  }));
+  if (!params.usePartialBatchResponses) {
+    throw new AggregateError(errors);
+  }
 
   params.logger.info(
     { batchItemFailures },