Worker Lambda & async tasks

WorkerFunction (actionstools-<env>-worker, handler lambda_handlers.worker_handler) is the only place background work runs. It replaces Zappa’s @task decorator and the separate prod2 deployment.

@async_task

from actionstools_app.tools.async_task import async_task

@async_task
@handle_async_errors
def notify_on_slack(slack_context, email, action_name, name):
    ...

Calling a decorated function:

  • With WORKER_FUNCTION_NAME set (both deployed functions): serialises {"task_path", "args", "kwargs"} with simplejson and calls lambda:Invoke with InvocationType="Event". The caller gets None back immediately.

  • Without it (tests, make run, CI): runs the function inline, synchronously.

The undecorated function stays reachable as fn.sync; the Worker calls that, so a task never re-queues itself.

  • Arguments must be JSON-serialisable — convert Decimal, datetime, enums and sets first.

  • The payload limit for asynchronous invokes is 256 KB.

  • There is no Flask request on the Worker. Read anything you need from request (for example get_auth_data()) before calling the task and pass it as an argument.

Keep @handle_async_errors inside @async_task: it initialises Sentry and logs the exception before re-raising.

Events the Worker accepts

worker_handler pushes a Flask app context, then dispatches on the event shape:

Event Source Runs

{"task_path": "actionstools_app…", "args": […​], "kwargs": {…​}}

@async_task (from either function)

The task’s .sync. Paths outside actionstools_app. and functions without .sync are rejected.

{"Records": [{"eventSource": "aws:dynamodb", …​}]}

TasksStream event source mapping

counters.task_db_trigger — Task counters

{"action": "remove_counters"}

RemoveCountersSchedule, rate(4 days), production only

counters.remove_counters

Anything else raises ValueError("Unrecognised worker event").

An async task event
{
  "task_path": "actionstools_app.tasks.notifications.notify_on_slack",
  "args": [
    {"login_id": "agent@payability.com", "task_url": "https://internal.payability.com/#/tasks?taskUuid=fcee792a-9797-4857-a948-02e4df2ed171", "number": 170, "title": "New task", "body": "<p>TEST</p>"},
    "assignee@payability.com",
    "assigned",
    "Assignee Name"
  ],
  "kwargs": {}
}

Failure semantics

Source On error

Async task

EventInvokeConfig.MaximumRetryAttempts: 0 — no retry (Zappa did not retry either). The error is in Sentry and the Worker’s CloudWatch logs.

Tasks stream

Retried up to 3 times, splitting the batch in half on each failure (BisectBatchOnFunctionError), then skipped.

Schedule

Not retried; the next run is 4 days later. Re-run by hand (below).

Running things by hand

# Re-run the counters cleanup
aws lambda invoke --function-name actionstools-prod-worker \
  --cli-binary-format raw-in-base64-out \
  --payload '{"action": "remove_counters"}' /dev/stdout

# Run any async task with explicit arguments
aws lambda invoke --function-name actionstools-stage-worker \
  --cli-binary-format raw-in-base64-out \
  --payload file://task.json /dev/stdout

Adding a task

  1. Write a module-level function in actionstools_app/… with JSON-friendly parameters.

  2. Decorate it @async_task (outer) and @handle_async_errors (inner).

  3. If it touches a new AWS resource, add the permission to both functions in template.yaml (the API may run it inline in tests, the Worker runs it in production).

  4. Add a test in tests/test_lambda_handlers.py style: call it with WORKER_FUNCTION_NAME unset to exercise the logic, and with a mocked Lambda client to check the payload.