Feature request: Concurrency control (avoidance of duplicated runs)

[pmarko1711]

Use case

Sometimes, I need to limit concurrency of a lambda function on the level of its payload. Concurrent invocations are fine but only if they have different payloads.

Background:

I already faced this need in two of my projects:

• The solution I came up with in my first project was very similar to what's behind Powertools' Idempotency - I created a decorator that before running my handler placed a lock item (with a ttl set) in a dynamodb table. The same decorator then released the lock after the handler finished successfully. In my case, I also wrote to the table any caught exceptions and "dynamic" parameters that I wanted to make available to subsequent invocations (with the same payload). This was before I was aware of Idempotency
• In my second project I needed the same kind of concurrency control, but without a need to preserve additional info in the underlying table. This time, I considered using Idempotency, but I couldn't as it always preserves function's output in its persistent storage layer (e.g. dynamodb, I believe here). It also does not release the cache/lock on success. I understand why - it offers idempotency, not a concurrency control. Still, it felt like I could (mis)use it for my usecase as well. Not wanting to hack it, I created another decorator ...

Solution/User Experience

Implement a decorator similar to @idempotent that would:

• prevent concurrent executions of a given lambda handler with the same payload making use of a lock placed in a dynamodb table
• decorator would release the lock on exceptions, but also if the decorated handler finishes successfully
• to safeguard against lambda timeouts it'd also set a ttl attribute (should correspond to the lambda function timeout)
• in contrast to @idempotent, it would not cache result and again, it'd release the lock on success

Similarly to @idempotent, payload should be either the entire event or its subset.

Optionally

• decorator could (perhaps optionally) write any exceptions caught to the underlying dynamodb table
• functionality to preserve parameters in the underlying dynamodb table, so that they are available to subsequent executions of the same lambda function with the same payload

Alternative solutions

Modify the @idempotent decorator to allow for switching off caching and releasing the lock on success. However, then the decorator can hardly be called idempotent ...

Acknowledgment

• This feature request meets Lambda Powertools Tenets
• Should this be considered in other Lambda Powertools languages? i.e. Java, TypeScript

[heitorlessa]

hey @pmarko1711 - thank you for a great feature request (details), I have an initial question: Have you considered Step Functions instead?

I've seen this a number of times where a Semaphore pattern would work out great, but instead done in an upstream component - a workflow of sorts.

I can guess a few reasons why you'd want to do in Lambda instead, so I'd love to hear more from you on why inside Lambda runtime vs an orchestration service like Step Functions.

Hope that makes sense

[pmarko1711]

Hi @heitorlessa , many thanks for replying. To be honest, I did not notice this pattern, so many thanks for sharing.

Indeed, it seems doable this way. Nevertheless; I still feel that in my case where I wanted to limit concurrency to 1 concurrent run (of the same payload of the same lambda; while allowing concurrent runs if different payload), it's just much easier with a decorator. Once I have it defined and I share it in my projects, activating such a concurrency control is just adding one or two lines of code. And I don't need to understand state machines (although those are arguably easier than a custom decorator :) )

Btw, my current logic looks this way:

from collections import namedtuple
from aws_lambda_powertools.middleware_factory import lambda_handler_decorator

MutexSettings = namedtuple(
    "MutexSettings",
    (
        "dyndb_resource", "dyndb_table",
        "attribute_partition_key", "attribute_sort_key", "attribute_ttl", "ttl",
        "attribute_mutex",  # (optional) name of the mutex attribute
        "mutex_locked",  # (optional) value to indicate a locked mutex
        "mutex_unlocked",  # (optional) value to indicate an unlocked mutex
        "suffix_partition_key",  # (optional) suffix to add to the partition key values for mutex items
        "suffix_sort_key",  # (optional) suffix to add to the sort key values for mutex items
    ),
    defaults=(
        "mutex",
        "LOCKED",
        "UNLOCKED",
        "",
        "",
    ),  # defaults are applied to the rightmost parameters
)


@lambda_handler_decorator()
def add_mutex(handler, event, context):

    # MUTEX_SETTINGS - MutexSettings  namedtuple instance 
    # here I also parse the event to get my partition_key and sort_key for which I want to
    # to set the lock
    # this could be replaced with something like idempontency_key

    # try locking (this raises an exception if already locked)
    # my own function that does a conditional update_item, setting a ttl item property
    acquire_mutex(partition_key, sort_key, MUTEX_SETTINGS)

    # run the handler and release the lock when the handler
    # - succeeds, but also when it fails
    try:
        return_val = handler(event, context)
        # my own function that releases the lock
        release_mutex(partition_key, sort_key, MUTEX_SETTINGS)
        return return_val

    except Exception as err:
        # my own function that releases the lock
        release_mutex(partition_key, sort_key, MUTEX_SETTINGS)
        raise err

@lambda_handler_decorator
def lambda_handler(event, context):
    pass

I can also provide my acquire_mutex, release_mutex (those are a bit larger), but they just do what I indicate in the comments above considering the settings object passed. And I was thinking to maybe make the decorator parametrizable, but this needs to be done smartly.

[heitorlessa]

Activating such a concurrency control is just adding one or two lines of code. And I don't need to understand state machines

yeah I figured that was the reason. It's about trade-off at the end. In simplistic terms: one one hand, this makes Lambda responsible for many things and may go out of sync if that spans to other functions, while delegating to an orchestrator requires domain knowledge and additional failure modes to be aware of (but solves semantics, concurrent invocations etc).

I'd love to hear whether this is useful for other customers too (I'll ping on Discord), so we can prioritize it.

If anything, I'd love to find a better umbrella for this feature, so we could incrementally improve operational posture for customers - think retry mechanisms (a thin wrapper for tenacity), a circuit breaker, and whatnot. If we work on a Concurrency Control utility, I'm unsure whether these falls under that too to ease discoverability.

I don't have any answers for now, just thinking out and loud.

[heitorlessa]

@pmarko1711 I'd love to have a call with you to learn a bit more your mutex use - I think there's a pattern across all new feature requests, RFCs, and customer interviews we've done recently... a call will be quicker than writing a narrative.

If you're interested and free next week and onwards, please ping us an email and we'll reply with a link to schedule a meeting on your earliest convenience: aws-lambda-powertools-feedback@amazon.com

Thank you!!

[walmsles]

🤔 hmmmm this seems to be a FIFO based requirement. Could FIFO SQS be used as a buffer here with message group id customised to group like messages? That way can also get batch execution safely within the lambda concurrency for efficiencies.

Although it also depends on other constraints and requirements regarding ordering of event processing, once you start using virtual FIFO queues ordering outside of groups goes away.

FIFO SQS message group id docs

I think we always need to look for the infrastructure to drive our code rather than add more actual code.

[pmarko1711]

Well, I'm not sure one can rely on SQS FIFO queues alone:

• (SQS FIFO) offers message deduplication, but only for 5 minute intervals. So a message triggers a Lambda that runs for 7 minutes and after 6 minutes the same message is sent to the queue, the function will be triggered again.
• It offers guarantee exactly-once processing, not exactly-once delivery. There are edge cases where a message can be delivered multiple times (and in my case thus trigger my lambda function).

https://ably.com/blog/sqs-fifo-queues-message-ordering-and-exactly-once-processing-guaranteed
https://www.spektor.dev/is-sqs-really-exactly-once/

It gets complex, I don't fully understand it, but what I get out of it is that if I want exactly once execution of my lambda function (with a given payload), I need to implement it in my lambda function.

(The group ids would help me to put the same payloads together, but I'd still be exposed to these problems.)

Please let me know if I'm wrong.

[heitorlessa]

Peter, quickly sharing the example we just discussed on the call to satisfy processing the same payload after X time (e.g., pre-aggregation use case) while limiting concurrency: https://github.com/awslabs/aws-lambda-powertools-python/blob/develop/tests/e2e/idempotency/handlers/ttl_cache_timeout_handler.py#L12

Contrary to DynamoDB TTL that is primarily used to remove DynamoDB items to save on cost + other use cases like archiving, Idempotency expires_after_seconds is used to signal when an existing COMPLETED should be updated to EXPIRED and completely ignored (allow the incoming transaction and record its output).

As for the output, whatever output your function returns we'll use it for later. But in your case, you can simply return an empty string to save on DynamoDB costs, since you're not gonna use it either way -- https://awslabs.github.io/aws-lambda-powertools-python/2.10.0/utilities/idempotency/#idempotent_function-decorator