Replay failed transactions

Depot Transactions create SQS messages when created. Transactions that fail will eventually be sent to a Dead Letter Queue (DLQ).

There are two main Transaction queues to look at for failed messages. You can replay messages back into the main Transaction queues by using a simple aws-sdk SQS client to capture messages from the DLQ and send them back to the correct input Transaction queue.

Main queues

The main Transaction queue is always named with convention:

• sdp-{env-id}-transaction (main Transaction entrypoint SQS queue)

The DLQ for the above main entrypoint queue is always named:

• sdp-{env-id}-dead-transaction

Replay messages

You can use the aws-sdk SQS client to receive messages from the DLQ (dead) queue and then forward them on to the main (transaction) queue for your environment. How you do this is up to you, but here is a fairly simple JavaScript example.

index.js

const { SQS }  = require('@aws-sdk/client-sqs');
const assert = require('assert');
const fs = require('fs');
const crypto = require("crypto");

const sqs = new SQS();

async function captureMessages(event) {

    const queueUrl = event.QueueUrl;
    assert.ok(queueUrl, "QueueUrl must be specified");
    const maxNumberOfMessages = event.MaxNumberOfMessages && parseInt(event.MaxNumberOfMessages) || 1;
    const visibilityTimeout = event.VisibilityTimeout && parseInt(event.VisibilityTimeout) || 30;
    const waitSeconds = event.WaitSeconds && parseInt(event.WaitSeconds) || 5;
    console.log(`Attempting to receiveMessage with MaxNumberOfMessages: ${maxNumberOfMessages}`);
    return await sqs.receiveMessage({
        QueueUrl: queueUrl,
        MaxNumberOfMessages: maxNumberOfMessages,
        WaitTimeSeconds: waitSeconds,
        VisibilityTimeout: visibilityTimeout
    }).then(response => {
        console.log("Got response: " + JSON.stringify(response));
        const messageFilename = crypto.randomUUID() + ".json";
        
        let data = {};
        if (response.Messages) {
            data = {
                Ok: true,
                HasMessages: true,
                Messages: response.Messages.map(v => { return {
                    ...v,
                    BodyJson: JSON.parse(v.Body)
                }})
            }
        } else {
            data = {
                Ok: true,
                HasMessages: false
            }
        }
        
        if (data.HasMessages) {
            data.Messages.forEach((msg) => {
                fs.writeFileSync('./msgs/' + messageFilename, JSON.stringify(msg.BodyJson));
            });
        }

        return data;
    });
}

async function forwardMessages(event, data) {

    const forwardQueueUrl = event.ForwardQueueUrl;
    assert.ok(forwardQueueUrl, "ForwardQueueUrl must be specified");
    const sendMessagePromises = [];
    if (data.HasMessages) {
        data.Messages.forEach((msg) => {
            
            const sendParams = {
                MessageBody: JSON.stringify(msg.BodyJson),
                QueueUrl: forwardQueueUrl
            };
            
            sendMessagePromises.push(
                sqs.sendMessage(sendParams)
            );
        });
    }

    console.log('replaying messages to ' + forwardQueueUrl);
    return await Promise.all(sendMessagePromises);
}

async function deleteMessages(event, data) {

    const queueUrl = event.QueueUrl;
    assert.ok(queueUrl, "QueueUrl must be specified");
    const deleteMessagesPromises = [];
    if (data.HasMessages) {
        data.Messages.forEach((msg) => {
            
            const deleteParams = {
                QueueUrl: queueUrl,
                ReceiptHandle: msg.ReceiptHandle
            };
            
            deleteMessagesPromises.push(
                sqs.deleteMessage(deleteParams)
            );
        });
    }

    console.log('deleting messages for original queue ' + queueUrl);
    return await Promise.all(deleteMessagesPromises);
}

async function handle(event) {

    const action = event.Action;
    if(action === "CaptureMessages") {
        return await captureMessages(event);
    } else if (action === "CaptureAndForwardMessages") {
        const data = await captureMessages(event);
        const forwardResult = await forwardMessages(event, data);
        console.log('forwarding complete', forwardResult);
        return await deleteMessages(event, data);
    } else {
        throw Error("No handler for action: " + action);
    }
}

exports.handler = async function(event) {

    console.log(event);
    const result = await handle(event);
    console.log(result);
    return result;
}

You would call this by importing the index.js module example above. For example:

example-capture-and-forward.js

const handler = require('./index');

const event = {
    Action: "CaptureAndForwardMessages",
    QueueUrl: "https://sqs.eu-west-1.amazonaws.com/123/sdp-19abba629acb-dead-transaction",
    ForwardQueueUrl: "https://sqs.eu-west-1.amazonaws.com/123/sdp-19abba629acb-transaction",
    MaxNumberOfMessages: 10,
    VisibilityTimeout: 30,
    WaitSeconds: 5
}

handler.handler(event).then((res) => {
    console.log('Completed', res);
});

Invoke the above example example-capture-and-forward.js by running it in node (being sure to use the correct AWS account named profile or context):

node example-capture-and-forward.js