Metadata-Version: 2.4
Name: celery-sqs-native-delay
Version: 0.1.0
Summary: Use SQS's native DelaySeconds for Celery ETA/countdown tasks instead of holding them in the worker's local timer.
Project-URL: Homepage, https://github.com/DHUKK/celery-sqs-native-delay
Project-URL: Repository, https://github.com/DHUKK/celery-sqs-native-delay
Project-URL: Issues, https://github.com/DHUKK/celery-sqs-native-delay/issues
Project-URL: Changelog, https://github.com/DHUKK/celery-sqs-native-delay/blob/main/CHANGELOG.md
Author-email: DHUKK <dennisjohnharrop@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: celery,delay,eta,kombu,sqs
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: Celery
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Requires-Dist: boto3>=1.34
Requires-Dist: celery>=5.3
Requires-Dist: kombu>=5.3
Description-Content-Type: text/markdown

# `celery-sqs-native-delay` ⏳️

[![CI](https://github.com/DHUKK/celery-sqs-native-delay/actions/workflows/ci.yml/badge.svg)](https://github.com/DHUKK/celery-sqs-native-delay/actions/workflows/ci.yml)
[![PyPI](https://img.shields.io/pypi/v/celery-sqs-native-delay.svg)](https://pypi.org/project/celery-sqs-native-delay/)
[![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)

`celery[sqs]` fetches every `ETA`/`countdown` task off the queue immediately and
holds it, unacked, in the worker's own memory until the `ETA` arrives. `SQS`
already has a native mechanism for this. This package makes Celery use it
by wrapping kombu's existing `SQS` transport.

## How it routes

```mermaid
flowchart LR
    A["task published<br/>with a delay"] --> B{".fifo queue?"}
    B -->|yes| C["celery's own<br/>ETA timer"]
    B -->|no| D{"delay ≤ 900s?"}
    D -->|yes| E["SQS native<br/>DelaySeconds"]
    D -->|no| F["EventBridge Scheduler<br/>one-time schedule"]
```

| Delay | Handled by |
| --- | --- |
| at or below `900s` | `SQS` native `DelaySeconds` |
| above `900s` | an `EventBridge` Scheduler one-time schedule |
| any delay on a `.fifo` queue | left alone, celery's own `ETA` timer |

> [!NOTE]
> `900s` is `SQS`'s hard limit on `DelaySeconds`, not a setting of this package.

## Install and configure

```
pip install celery-sqs-native-delay
```

```python
app = Celery(
    "myapp",
    broker="sqs://",
    broker_transport="celery_sqs_native_delay:Transport",
    broker_transport_options={
        "eventbridge_role_arn": "arn:aws:iam::123456789012:role/celery-scheduler",
    },
)
```

| `broker_transport_options` key | Required | Meaning |
| --- | --- | --- |
| `eventbridge_role_arn` | yes | role `EventBridge` Scheduler assumes to call `SendMessage` |
| `eventbridge_dead_letter_arn` | no, but recommended | `DeadLetterConfig` on every schedule |
| `eventbridge_retry_policy` | no | `RetryPolicy` on every schedule |
| `eventbridge_group_name` | no | schedule group |

## Requirements and limitations

- **Celery protocol 2 is required.** Protocol 2 has been celery's default
  since `4.0` (2016), so almost every app already meets this. Protocol 2 puts
  `eta` in the message headers. Protocol 1 puts it in the body, which this
  package never inspects, so a protocol 1 delay is invisible from here and
  falls back silently to celery's own local timer.
- **FIFO queues get no native delay.** `SQS` has no per-message `DelaySeconds`
  on a `.fifo` queue, and `EventBridge` Scheduler can't be used either, since
  separate schedule invocations have no ordering guarantee relative to each
  other. A delayed publish to a `.fifo` queue falls back entirely to
  celery's own local `ETA` timer, exactly as if this package weren't
  installed.
    - The way to get delay on a `FIFO` queue without any of that is a
      queue-level `DelaySeconds` attribute set at queue creation. It applies
      uniformly to every message and preserves ordering.

## IAM

Three things need permissions: the role in `eventbridge_role_arn`, that
role again for delivery, and the publisher itself.

**1. The role trusts `EventBridge` Scheduler to assume it:**

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": { "Service": "scheduler.amazonaws.com" },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringEquals": { "aws:SourceAccount": "123456789012" },
        "ArnLike": {
          "aws:SourceArn": "arn:aws:scheduler:us-east-1:123456789012:schedule/*/*"
        }
      }
    }
  ]
}
```

**2. The role can deliver to your queue(s):**

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "sqs:SendMessage",
      "Resource": "arn:aws:sqs:us-east-1:123456789012:myqueue"
    }
  ]
}
```

**3. The publisher can create schedules and pass the role to `EventBridge`:**

```json
{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["scheduler:CreateSchedule", "sqs:GetQueueAttributes"],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": "iam:PassRole",
      "Resource": "arn:aws:iam::123456789012:role/celery-scheduler"
    }
  ]
}
```

## Failure and recovery

A `CreateSchedule` failure (throttling, a missing role, etc) raises
out of `Channel._put`, which propagates up through `producer.publish`, so
celery's own `task_publish_retry` covers it with no extra handling needed
from this package.

A schedule that *fires* but cannot deliver, for instance a deleted queue, a
revoked role permission, or a KMS error on an encrypted queue, goes to
`eventbridge_dead_letter_arn` if you set one. Setting it is recommended:
without it, a schedule that fails to deliver leaves nothing behind, because
`ActionAfterCompletion='DELETE'` removes the schedule once it has run.

## Retries

Celery reuses one task id across every retry, but "The rule" above is
applied fresh on each publish. A retry with a growing backoff can start
under `900s` on native `DelaySeconds`, then move to `EventBridge` Scheduler
once a later retry's countdown passes `900s`, with no special handling on
your part:

```python
@app.task(bind=True, max_retries=5)
def my_task(self):
    try:
        ...
    except Exception as exc:
        countdown = 300 * (2**self.request.retries)
        raise self.retry(exc=exc, countdown=countdown)
```

## Quotas

Every delay above `900s` costs one `EventBridge` Scheduler `CreateSchedule`
call and one invocation. Check your account's `EventBridge` Scheduler quotas
(schedule count, `CreateSchedule` rate, invocation rate) against your
expected volume before relying on this path at scale.

## Development

```
make install            # uv sync --dev
make test               # unit tests, no broker needed
make up                 # start a local ministack (https://ministack.org)
make test-integration   # integration tests against that ministack
make test-all           # integration + unit tests
make down
make lint
```

## License

MIT, see [`LICENSE`](LICENSE).
