CUMULUS-4300 - Implement Rate-Limited Lambda Dispatcher for SQS Queue (Max 5 Messages/Second) #4166

mikedorfman · 2025-12-11T18:42:26Z

Summary: These changes introduce a new rate-limited consumer class in the Node/TypeScript code to control how many executions are submitted per second across multiple queues - helping improve and smooth out step function submission.

Addresses CUMULUS-4300: Implement Rate-Limited Lambda Dispatcher for SQS Queue (Max 5 Messages/Second)

Changes

Node/TypeScript:
- Created a new ConsumerRateLimited class that is able to submit executions at a specified, even maximum rate as defined by rateLimitPerSecond. In order to enforce this limit across all throttled queues, this class accepts a list of queue URLs instead of a single throttled queue URL. Unlike its non-rate-limited counterpart, to simplify configuration, this new class does not limit the number of messages staged - that can now be indirectly controlled by increasing or decreasing the rate.
- Added calls to the new ConsumerRateLimited class in sf-starter.js in the handleRateLimitedEvent function. This uses the incrementAndDispatch dispatcher.
Terraform:
- Added a new Lambda named "sqs2sfThrottleRateLimited" that can be called with a list of queueURLs in an EventBridge scheduled rule.
- Added sqs2sfThrottleRateLimited_lambda_function_arn outputs to both ingest and cumulus modules.

PR Checklist

Update CHANGELOG
Unit tests
Ad-hoc testing - Deploy changes and test manually
Integration tests

📝 Note:
For most pull requests, please Squash and merge to maintain a clean and readable commit history.

Reworked rate limiting code in the Consumer

…n the lambda duration time, which is also configurable Updated default behavior to not include a message limit but instead stage as many messages as possible within the time provided Exposed sf_start_limit and sf_start_rate variables

Updated default rateLimitPerSecond max to match experimental values that correspond to a throughput of 5/second

Reverted previously-changed handleEvent function changes Updated wait time to 5 seconds and time buffer to 1 second Fixed a bug where we were using the wrong compare operator

etcart · 2025-12-11T21:56:24Z

packages/ingest/src/consumerRateLimited.ts

+    for (const [message, queueUrl] of messagesWithQueueUrls) {
+      const waitTime = 1000 / this.rateLimitPerSecond;
+      log.debug(`Waiting for ${waitTime} ms`);
+      await sleep(waitTime); // eslint-disable-line no-await-in-loop


Suggested change

await sleep(waitTime); // eslint-disable-line no-await-in-loop

await sleep(waitTime);

So - this is the correct line, the issue is @mikedorfman pulled in the file ignore wholesale from my suggestions/CI run commit - that's not right, we should except individual blocks. My bad for not calling it out. 8380273

Cool, sounds good - thanks for the clarification!

etcart · 2025-12-11T21:59:32Z

packages/ingest/src/consumerRateLimited.ts

+      log.debug(`Waiting for ${waitTime} ms`);
+      await sleep(waitTime); // eslint-disable-line no-await-in-loop
+      if (await this.processMessage(message, fn, queueUrl)) {
+        counter++;


Suggested change

counter++;

counter += 1;

Updated in the next commit

packages/ingest/test/test-consumerRateLimited.js

Jkovarik · 2025-12-11T22:24:10Z

packages/ingest/src/consumerRateLimited.ts

+export class ConsumerRateLimited {
+  private readonly deleteProcessedMessage: boolean;
+  private readonly queueUrls: string[];
+  private readonly timeRemainingFunc: () => number;
+  private readonly visibilityTimeout: number;
+  private readonly rateLimitPerSecond: number;
+  private readonly timeBuffer: number;
+  private readonly messageLimitPerFetch: number;
+  private readonly waitTime: number;
+
+  constructor({
+    queueUrls,


Nit: consider commenting all of the class params in the interface, using optional parameters for the params that are being set to static ingegers and doing something like

e.g.

/** * URLs of the SQS queues to poll. */ queueUrls: string[];

and so on in the interface. Totally get the original consumer didn't have this.

Updated in the next commit

Oh, actually @Jkovarik can you clarify what you mean by using optional parameters for the params that are being set to static integers? I was thinking you were saying;

private readonly timeBuffer?: number; private readonly messageLimitPerFetch?: number; private readonly waitTime?: number;

But I think I'm misinterpreting since they're always set to values so aren't optional.

packages/ingest/src/consumerRateLimited.ts

Jkovarik · 2025-12-11T22:27:41Z

packages/ingest/src/consumerRateLimited.ts

+    // we'll immediately start fetching the next batch while processing the
+    // current one
+
+    let messages = await Promise.all(


Nit: We're repeating this 3x, consider abstracting to a helper/private method

Creating a fetchMessagesFromAllQueues method in the next commit.

Jkovarik · 2025-12-11T22:36:30Z

packages/api/lambdas/sf-starter.js

 }

+/**
+ * This is an SQS queue consumer.


Jkovarik · 2025-12-11T22:47:38Z

packages/api/lambdas/sf-starter.js

+ * in the event object. It is a rate-limited version of the throttled consumer.
+ *
+ *
+ * @param {Object} event - lambda input message


Something that's entirely not-obvious re: the state of the Core codebase, we have an ADR that the team/repo is trying to enable ts-check on any modified .js file and make it correct for the modified code. We should consider doing so for this docstring and avoid Object as a valid type except in specific exceptions.

Ah, I see - that makes sense. Updated in the next commit.

Updated with additional PR feedback

Jkovarik · 2025-12-12T19:57:28Z

packages/ingest/src/consumerRateLimited.ts

+   */
+  timeRemainingFunc: () => number;
+  /**
+   * The visibility timeout used when fetching messages from the SQS queues.


Nit:

Suggested change

* The visibility timeout used when fetching messages from the SQS queues.

* The visibility timeout in milleseconds used when fetching messages from the SQS queues.

Updated in the next commit

Jkovarik · 2025-12-12T20:49:13Z

tf-modules/ingest/lambda-functions.tf

+  role             = var.lambda_processing_role_arn
+  runtime          = "nodejs20.x"
+  # This timeout must match the cadence of the event bridge rule that triggers it
+  timeout          = 60


Had a brief conversation offline with @mikedorfman re: this, I think there's a potential timing issue with this timeout being too close to the event loop in clearing sempahores meaning the lambda could end without clearing semaphore. We talked through it and it's probably the right call to extend the timeout and make sure the event loop is configured such that it stops dispatching based on the EB periodicity but the lambda has a much longer timeout to allow things like semaphores to clear.

Noting here to track changes/etc.

packages/api/lambdas/sf-starter.js

Jkovarik · 2025-12-12T20:56:55Z

packages/api/lambdas/sf-starter.js

 * @returns {Promise} - AWS SF Start Execution response
 */
-function dispatch(queueUrl, message) {
+async function dispatch(queueUrl, message) {


Just to clarify - this (and awaiting sfn().StartExecution) was done for readability?

I'm trying to remember why I had update this - I believe it was my initial confusion about async functions. Given we await the dispatch function when it's called, does it make sense to update this to an async function? I don't think including or excluding this change impacts my update.

Mostly just making sure I understood why the change was made/there wasn't some side-effect somewhere I was missing.

Functionally awaiting here and making this async for something else to await is more explicit versus having a non-async function that passes back an awaitable promise, so I think it's a net good change.

Jkovarik · 2025-12-12T20:58:01Z

packages/ingest/src/consumerRateLimited.ts

@@ -0,0 +1,161 @@
+/* eslint-disable no-await-in-loop */


This should be removed (I added it to get the local commit hooks to pass without cluttering pre-edit), apologies

Done in next commit

@mikedorfman I don't see this as removed - were you saying you have work in-flight you're going to push up?

Jkovarik · 2025-12-12T21:01:35Z

packages/ingest/test/test-consumerRateLimited.js

+
+test.beforeEach(() => {
+  testConsumer = new ConsumerRateLimited({
+    queueUrls: [fakeQueueName],


We should probably test multiple queues.... although with the stubbing here that might be brittle... and it's stubbed like that in the OG code.

We have the option to use localstack instead and actually test consuming messages, however there's a lot of overhead there.

More generally: I think we either need to make these units more comprehensive or write a integration test to validate this is working.

I'm in the process of adding a test that exercises the sf-starter handleRateLimitedEvent functionality, hopefully that can cover this case (and I might add a unit test with multiple queues in anyway as well).

packages/api/lambdas/sf-starter.js

Jkovarik

This looks really good! Just a couple of changes to consider/discuss other than minor nits/lint type stuff - the race/timeout condition concern and an integration or unit test update.

Oh - and CHANGELOG entry please!

Let me know what you think!

- Modified the time remaining function to use a value separate from the lambda timeout

Additional updates per PR comments

Jkovarik · 2025-12-16T16:40:11Z

packages/ingest/src/consumerRateLimited.ts

+          `No messages fetched, waiting ${this.waitTime} ms before retrying`
+        );
+        await sleep(this.waitTime);
+        messages = await this.fetchMessagesFromAllQueues();


Nit: re-reading this, I think this could lead to some lag in low-message volume scenarios. The sleep isn't validating the time remaining before sleeping, so in theory you could pull messages you already know you can't process due to the while loop. The cost here is just the visibility timeout. On the fence about that being a nit.

Ah, I see. Looking at it closer myself, I think we'd always end up with messages that were fetched but not processed. since we're trying to fetchMessagesFromAllQueues while we're running the last message processing. I think if we add a processMessages call outside the while loop, that would clean up the last batch pulled from fetchMessagesFromAllQueues.

I'll add a unit test to verify this and cover this scenario.

Reworked consumerRateLimited .consume method to avoid fetching messages we won't process Addressed other PR feedback

…stimation of the time remaining

…da timeout test

Dismissing as my initial concerns were addressed and/or updated and re-submitted for review. Integration tests should be reviewed.

Jkovarik · 2025-12-22T13:29:11Z

@mikedorfman thanks for all your work on this! I dismissed my review as I believe you addressed my initial comments, but could not review the integration test updates on Friday.

GitHub Actions and others added 13 commits November 19, 2025 12:35

Updated sf-starter to enforce a wait time for throttled queues

6c4440c

Updated default timeout for sfstarter to 60 seconds

035e306

Reworked rate limiting code in the Consumer

Updated sqs2sfThrottle default timeout to match the eventbridge cadence

42d9f6e

Updated default rateLimitPerSecond max to match experimental values that correspond to a throughput of 5/second

Reworked rate-limited consumer code updates

92846a0

Added some resources needed for isolated deployment of sqs2sfThrottle

88b4ea6

Cleaned up changes not intended for this PR

985f4a6

Updated verbose logs to debug level

d67ef57

Reverted previously-changed handleEvent function changes Updated wait time to 5 seconds and time buffer to 1 second Fixed a bug where we were using the wrong compare operator

Added tests for new throttled queue configuration

c2d0c33

Reverting further changes

9c52d38

Lint/syntax fixes for PR suggestions

8380273

Updates per PR comments

d22c432

Added a newline at the end of the test file

871aa95

Jkovarik added the in review label Dec 11, 2025

etcart reviewed Dec 11, 2025

View reviewed changes

packages/ingest/test/test-consumerRateLimited.js Show resolved Hide resolved

Jkovarik reviewed Dec 11, 2025

View reviewed changes

packages/api/lambdas/sf-starter.js

}

/**

* This is an SQS queue consumer.

Copy link

Member

Jkovarik Dec 11, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

😍

Jkovarik reviewed Dec 11, 2025

View reviewed changes

Added additional typing and typedefs throughout

f66e1ea

Updated with additional PR feedback

Jkovarik reviewed Dec 12, 2025

View reviewed changes

packages/api/lambdas/sf-starter.js Outdated Show resolved Hide resolved

Jkovarik reviewed Dec 12, 2025

View reviewed changes

packages/api/lambdas/sf-starter.js Show resolved Hide resolved

Jkovarik previously requested changes Dec 12, 2025

View reviewed changes

Jkovarik removed the in review label Dec 12, 2025

Jkovarik added the pr feedback label Dec 12, 2025

mikedorfman added 7 commits December 15, 2025 06:55

Updates per PR feedback:

9a3baf5

- Modified the time remaining function to use a value separate from the lambda timeout

Updates per PR review

c0d4da1

Updated ChangeLog

3eec1b7

Additional updates per PR comments

Merge branch 'master' into md/CUMULUS-4300

343a6ec

Added a Changelog entry

f4326d2

Fixed linting issues

c804c5b

More linting fixes

60696b9

Jkovarik reviewed Dec 16, 2025

View reviewed changes

mikedorfman added 6 commits December 18, 2025 15:13

Added integration tests to exercise new rate-limiting functionality

a702156

Reworked consumerRateLimited .consume method to avoid fetching messages we won't process Addressed other PR feedback

Updated the exit buffer time calculation to include a more accurate e…

f48842f

…stimation of the time remaining

updated tests based on updates to the sf-starter

3d47508

updated test

eb3334e

Uncommented accidentally-commented-out tests - d'oh!

2c2148d

Removed flawed handleThrottledRateLimitedEvent ends prior to the lamb…

bbb3972

…da timeout test

Jkovarik added needs review and removed pr feedback labels Dec 22, 2025

	await sleep(waitTime); // eslint-disable-line no-await-in-loop
	await sleep(waitTime);

	* The visibility timeout used when fetching messages from the SQS queues.
	* The visibility timeout in milleseconds used when fetching messages from the SQS queues.

CUMULUS-4300 - Implement Rate-Limited Lambda Dispatcher for SQS Queue (Max 5 Messages/Second) #4166

Are you sure you want to change the base?

CUMULUS-4300 - Implement Rate-Limited Lambda Dispatcher for SQS Queue (Max 5 Messages/Second) #4166

Uh oh!

Conversation

mikedorfman commented Dec 11, 2025

Changes

PR Checklist

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Jkovarik Dec 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

etcart Dec 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Jkovarik Dec 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Jkovarik left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Jkovarik commented Dec 22, 2025

Uh oh!

Reviewers

Assignees

Labels

Jkovarik Dec 11, 2025 •

edited

Loading

etcart Dec 11, 2025 •

edited

Loading

Jkovarik Dec 12, 2025 •

edited

Loading

Jkovarik left a comment •

edited

Loading