fix: move publishRetryMessage to end of loop and skip invalid messages

Per reviewer feedback, publishRetryMessage is now called at the end of the
processing loop for each message that is NOT marked as invalid. This prevents
duplicate retry messages for the same event, since invalid messages already go
back to the SQS queue for retry.

Key changes:
- Track validMessagesForRetry separately from all messages
- Only messages that pass job queued check are added to validMessagesForRetry
- publishRetryMessage is called after runner creation, not before
- Messages marked as invalid (e.g., max runners reached, creation failed) are
  excluded from retry message publishing

Tests updated to reflect new behavior:
- publishRetryMessage is called AFTER runner creation
- Messages marked invalid do not get retry messages published
- All test scenarios updated with proper mock runner creation

Co-authored-by: npalm <11609620+npalm@users.noreply.github.com>

Author: Copilot

lambdas/functions/control-plane/src/scale-runners/scale-up.test.ts

@@ -1698,6 +1698,7 @@ describe('Retry mechanism tests', () => {
 
   it('calls publishRetryMessage for each valid message when job is queued', async () => {
     const messages = createTestMessages(3);
+    mockCreateRunner.mockResolvedValue(['i-12345', 'i-67890', 'i-abcdef']); // Create all requested runners
 
     await scaleUpModule.scaleUp(messages);
 
@@ -1749,15 +1750,16 @@ describe('Retry mechanism tests', () => {
     );
   });
 
-  it('calls publishRetryMessage even when maximum runners is reached', async () => {
+  it('does not call publishRetryMessage when maximum runners is reached and messages are marked invalid', async () => {
     process.env.RUNNERS_MAXIMUM_COUNT = '0'; // No runners can be created
 
     const messages = createTestMessages(2);
 
     await scaleUpModule.scaleUp(messages);
 
-    // publishRetryMessage should still be called even though no runners will be created
-    expect(mockPublishRetryMessage).toHaveBeenCalledTimes(2);
+    // publishRetryMessage should NOT be called because messages are marked as invalid
+    // Invalid messages go back to the SQS queue and will be retried there
+    expect(mockPublishRetryMessage).not.toHaveBeenCalled();
     expect(createRunner).not.toHaveBeenCalled();
   });
 
@@ -1781,6 +1783,7 @@ describe('Retry mechanism tests', () => {
 
   it('calls publishRetryMessage when ENABLE_JOB_QUEUED_CHECK is false', async () => {
     process.env.ENABLE_JOB_QUEUED_CHECK = 'false';
+    mockCreateRunner.mockResolvedValue(['i-12345', 'i-67890']); // Create all requested runners
 
     const messages = createTestMessages(2);
 
@@ -1792,6 +1795,7 @@ describe('Retry mechanism tests', () => {
   });
 
   it('calls publishRetryMessage for each message in a multi-runner scenario', async () => {
+    mockCreateRunner.mockResolvedValue(['i-12345', 'i-67890', 'i-abcdef', 'i-11111', 'i-22222']); // Create all requested runners
     const messages = createTestMessages(5);
 
     await scaleUpModule.scaleUp(messages);
@@ -1808,8 +1812,9 @@ describe('Retry mechanism tests', () => {
     });
   });
 
-  it('calls publishRetryMessage before runner creation', async () => {
+  it('calls publishRetryMessage after runner creation', async () => {
     const messages = createTestMessages(1);
+    mockCreateRunner.mockResolvedValue(['i-12345']); // Create the requested runner
 
     const callOrder: string[] = [];
     mockPublishRetryMessage.mockImplementation(() => {
@@ -1823,7 +1828,7 @@ describe('Retry mechanism tests', () => {
 
     await scaleUpModule.scaleUp(messages);
 
-    expect(callOrder).toEqual(['publishRetryMessage', 'createRunner']);
+    expect(callOrder).toEqual(['createRunner', 'publishRetryMessage']);
   });
 });
 

lambdas/functions/control-plane/src/scale-runners/scale-up.ts

@@ -339,6 +339,7 @@ export async function scaleUp(payloads: ActionRequestMessageSQS[]): Promise<stri
   for (const [group, { githubInstallationClient, messages }] of validMessages.entries()) {
     // Work out how much we want to scale up by.
     let scaleUp = 0;
+    const validMessagesForRetry: ActionRequestMessageSQS[] = [];
 
     for (const message of messages) {
       const messageLogger = logger.createChild({
@@ -357,7 +358,7 @@ export async function scaleUp(payloads: ActionRequestMessageSQS[]): Promise<stri
       }
 
       scaleUp++;
-      await publishRetryMessage(message as ActionRequestMessageRetry);
+      validMessagesForRetry.push(message);
     }
 
     if (scaleUp === 0) {
@@ -398,6 +399,12 @@ export async function scaleUp(payloads: ActionRequestMessageSQS[]): Promise<stri
 
       // No runners will be created, so skip calling the EC2 API.
       if (newRunners <= 0) {
+        // Publish retry messages for all remaining messages that are not marked as invalid
+        for (const message of validMessagesForRetry) {
+          if (!invalidMessages.includes(message.messageId)) {
+            await publishRetryMessage(message as ActionRequestMessageRetry);
+          }
+        }
         continue;
       }
     }
@@ -451,6 +458,13 @@ export async function scaleUp(payloads: ActionRequestMessageSQS[]): Promise<stri
 
       invalidMessages.push(...messages.slice(0, failedInstanceCount).map(({ messageId }) => messageId));
     }
+
+    // Publish retry messages for all messages that are not marked as invalid
+    for (const message of validMessagesForRetry) {
+      if (!invalidMessages.includes(message.messageId)) {
+        await publishRetryMessage(message as ActionRequestMessageRetry);
+      }
+    }
   }
 
   return invalidMessages;