Limited sqs size

CHANGELOG.md

@@ -65,6 +65,8 @@ each Cumulus version between your current version and v19.1.0 as normal.
 
 ### Changed
 
+- **CUMULUS-3678**
+  - sqs messages truncated to max sqs length if necessary
 - **CUMULUS-3928**
   - updated publish scripting to use [email protected] for user email
   - updated publish scripting to use esm over common import of latest-version

packages/aws-client/src/SQS.ts

@@ -18,7 +18,9 @@ import {
   Message,
   QueueAttributeName,
   ReceiveMessageCommand,
-  SendMessageCommand } from '@aws-sdk/client-sqs';
+  SendMessageCommand,
+  SendMessageCommandOutput,
+} from '@aws-sdk/client-sqs';
 
 import { StepFunctionEventBridgeEvent } from './Lambda';
 import { sqs } from './services';
@@ -87,20 +89,45 @@ export const getQueueAttributes = async (queueName: string) => {
   };
 };
 
+/**
+ * Ensure that a string does not overrun SQS max body size of 262144 bytes
+ */
+export const limitSQSMessageLength = (str: string) => {
+  const maxSize = 262144;
+  // because this string can be a mix of
+  // if there is a chance this is overflowing, do a more expensive check and fix
+  if (str.length <= maxSize / 2) {
+    return str;
+  }
+
+  const encoded = (new TextEncoder()).encode(str);
+  if (encoded.length <= maxSize) {
+    return str;
+  }
+  const warningString = '...TruncatedForLength';
+  // In the specific edge case that a unicode (2 byte) character is split
+  // TextDecoder decoding pads it with 2 hidden bytes
+  const finalStr = new TextDecoder().decode(encoded.slice(0, maxSize - warningString.length - 2));
+  return `${finalStr}${warningString}`;
+};
+
 /**
  * Send a message to AWS SQS
  **/
 export const sendSQSMessage = (
   queueUrl: string,
   message: string | object,
   logOverride: Logger | undefined = undefined
-) => {
+): Promise<SendMessageCommandOutput> => {
   const logger = logOverride || log;
   let messageBody;
+  console.log(messageBody);
   if (isString(message)) messageBody = message;
   else if (isObject(message)) messageBody = JSON.stringify(message);
   else throw new Error('body type is not accepted');
 
+  messageBody = limitSQSMessageLength(messageBody);
+
   const command = new SendMessageCommand({
     MessageBody: messageBody,
     QueueUrl: queueUrl,

packages/aws-client/tests/test-SQS.js

@@ -12,6 +12,8 @@ const {
   sqsQueueExists,
   sendSQSMessage,
   isSQSRecordLike,
+  receiveSQSMessages,
+  limitSQSMessageLength,
 } = require('../SQS');
 
 const randomString = () => cryptoRandomString({ length: 10 });
@@ -100,3 +102,46 @@ test('isSQSRecordLike filters correctly for sqs record shape', (t) => {
   body *should* be a string form json object,
   but strictly checking is not compatible with the current use-case*/
 });
+
+test('sendSQSMessage truncates oversized messages safely', async (t) => {
+  const queueName = randomString();
+  const queueUrl = await createQueue(queueName);
+  const maxLength = 262144;
+  const overflowMessage = '0'.repeat(maxLength + 2);
+  await sendSQSMessage(queueUrl, overflowMessage);
+
+  const recievedMessage = await receiveSQSMessages(queueUrl, {});
+  const messageBody = recievedMessage[0].Body;
+  t.true(messageBody.endsWith('...TruncatedForLength'));
+  t.true(messageBody.length <= maxLength);
+});
+
+test('limitSQSMessageLength truncates unicode messages of greater than maximum byte size', (t) => {
+  const maxLength = 262144;
+  const overflowMessageUnicodeMessage = 'è'.repeat(maxLength / 2 + 20);
+  const massagedMessage = limitSQSMessageLength(overflowMessageUnicodeMessage);
+
+  t.true(massagedMessage.endsWith('...TruncatedForLength'));
+  t.true(massagedMessage.length <= maxLength);
+
+  const overflowMessageMixedMessage = 'èa'.repeat(maxLength / 2 + 20);
+  const massagedMixedMessage = limitSQSMessageLength(overflowMessageMixedMessage);
+
+  t.true(massagedMixedMessage.endsWith('...TruncatedForLength'));
+  t.true(massagedMixedMessage.length <= maxLength);
+});
+
+test('limitSQSMessageLength does not truncate messages appropriate for sqs to handle', (t) => {
+  const maxLength = 262144;
+  let underflowMessage = '0'.repeat(maxLength);
+  t.is(limitSQSMessageLength(underflowMessage), underflowMessage);
+
+  underflowMessage = 'ř'.repeat(maxLength / 2);
+  t.is(limitSQSMessageLength(underflowMessage), underflowMessage);
+
+  underflowMessage = 'a'.repeat(maxLength / 2);
+  t.is(limitSQSMessageLength(underflowMessage), underflowMessage);
+
+  underflowMessage = 'abcd';
+  t.is(limitSQSMessageLength(underflowMessage), underflowMessage);
+});