[MehrazRumman]

Description

Channel.basic_reject(delivery_tag, requeue=False) on the SQS transport only released the delivery locally. The message stayed in SQS, so it came back once its visibility timeout expired and was received again and again forever.

This PR adds basic_reject to the SQS Channel. When a message is rejected without requeue, it is now deleted from SQS (DeleteMessage with its receipt handle). This is the same fix proposed in #2711, with these differences:

• The backoff policy keeps working. Celery calls reject(requeue=False) when a task fails, and that is when backoff_policy / backoff_tasks is documented to apply. Messages whose task is listed in backoff_tasks for the queue are still left in SQS and get their visibility timeout changed as before, so they are retried. A new QoS.is_backoff_task() helper decides this.
• requeue=True is unchanged. The backoff path is not moved behind requeue=True.
• The delivery is always released. It is released in a finally, even if DeleteMessage raises, so a failed delete cannot hold a prefetch slot.
• Missing SQS info is not an error. If the delivery has no sqs_message/sqs_queue info, or the tag is unknown, nothing is deleted and the call falls back to the default reject.
• Docs: the SQS transport docstring now describes how reject behaves.

When a rejected message is kept

The message is left in the queue, as before this PR, when:

• its task is listed in backoff_tasks, so the backoff policy can retry it;
• the queue has a redrive policy, so SQS moves the message to the dead-letter queue once it has been received maxReceiveCount times. The policy is read once per queue with GetQueueAttributes, and if that call fails the message is kept;
• DeleteMessage fails with a ClientError, for example an expired receipt handle. The error is logged instead of raised, so the worker keeps running, and SQS redelivers the message.

Tests

The new unit tests cover every added line:

• Reject without requeue deletes the message, with or without a routing_key.
• Reject with requeue does not delete.
• A backoff task rejected without requeue gets its visibility timeout changed instead of being deleted.
• Missing sqs_message / sqs_queue, and unknown delivery tags.
• The delivery is released when DeleteMessage raises, and a ClientError is logged rather than raised.
• Queues with a redrive policy keep the message; the policy is read once per queue; a failed GetQueueAttributes keeps the message.
• QoS.is_backoff_task() for listed, unlisted and missing tasks, and for queues with no backoff config.

Integration tests (t/integration/test_sqs.py) reject a message on a queue with a 1 second visibility timeout and check that it is not redelivered, or that it is redelivered when the queue has a redrive policy.

Related: #2711,

Fixes: #2421

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 85.72%. Comparing base (cf604c2) to head (6dc0261).

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2784      +/-   ##
==========================================
+ Coverage   85.63%   85.72%   +0.08%     
==========================================
  Files          80       80              
  Lines       11543    11592      +49     
  Branches     1394     1401       +7     
==========================================
+ Hits         9885     9937      +52     
+ Misses       1436     1434       -2     
+ Partials      222      221       -1     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

[auvipy]

@ishan-1010 @ChristianMurphy can you check and review this?

[auvipy]

Is the change already covered by existing integration tests?

[auvipy]

@ChristianMurphy can you check this?

[ChristianMurphy]

Sure. I tested 48ff39d with Celery main and a moto SQS server. A message rejected without requeue is now deleted, so SQS no longer redelivers it in a loop.

Deleting it breaks SQS dead-letter queues. With task_acks_late=True and task_acks_on_failure_or_timeout=False, Celery rejects every failed task without requeue. Celery added task_acks_on_failure_or_timeout in 4.3 for SQS users who let SQS retry a failed task and then move it to a dead-letter queue (changelog). With maxReceiveCount: 3, a failing task runs 3 times on main and then moves to the dead-letter queue. On this branch it runs once, and the message is gone.

To answer the opt-in question in the description: I'd skip the delete when the queue has a redrive policy. kombu can read the policy with GetQueueAttributes(AttributeNames=['RedrivePolicy']), which _size() already calls. It only needs to ask once per queue. If AWS denies that call, not deleting seems safest. An opt-in option would also work, but users who don't know about it would still see the loop. Either way, the new docstring paragraph should say what happens on a queue with a redrive policy.

A failed delete stops the Celery worker. basic_reject lets botocore's ClientError through, and Celery's reject() only catches the transport's connection errors. When DeleteMessage failed with the expired receipt handle error, the worker logged Unrecoverable error and exited. basic_ack already catches ClientError: it drops the delivery for a stale receipt handle and rejects it for other errors (SQS/init.py:958-973). Could basic_reject do the same?

On auvipy's integration test question: no. test_reject_no_requeue passes on main too, because it checks while the 1800 second visibility timeout still hides the message. The test below uses a 1 second timeout. It fails on main and passes on this branch. I ran it against moto, not GoAws.

The new tests test_basic_reject_unknown_delivery_tag, test_is_backoff_task and test_is_backoff_task_without_backoff_config leave a mock in qos._delivered. At exit, kombu can't pickle it, prints EMERGENCY DUMP STATE TO FILE, and writes a file to /tmp for each test. Clearing channel.qos._delivered at the end of each test fixes it.

The test

@pytest.mark.env('sqs')
def test_reject_no_requeue_is_not_redelivered(connection):
    conn = connection.clone(transport_options={
        **connection.transport_options, 'visibility_timeout': 1,
    })
    with conn, conn.SimpleQueue('reject_no_requeue_vt') as queue:
        queue.put({'Hello': 'World'})
        queue.get(timeout=5).reject(requeue=False)
        time.sleep(2)  # longer than the visibility timeout
        with pytest.raises(queue.Empty):
            queue.get(timeout=2)

Worker runs

One solo worker on Celery main, with task_acks_late=True and task_acks_on_failure_or_timeout=False. The queue has VisibilityTimeout=2, and each run lasts 15 seconds. fails raises ValueError, and rejects raises Reject(requeue=False).

kombu	Task	Redrive policy	Task runs	Left in queue	In dead-letter queue
main	fails	maxReceiveCount: 3	3	0	1
main	fails	none	6	1	0
main	rejects	maxReceiveCount: 3	3	0	1
main	rejects	none	7	1	0
48ff39d	fails	maxReceiveCount: 3	1	0	0
48ff39d	fails	none	1	0	0
48ff39d	rejects	maxReceiveCount: 3	1	0	0
48ff39d	rejects	none	1	0	0

I patched every DeleteMessage call to fail with the expired receipt handle error. On 48ff39d, the worker exited with code 1 after fails or rejects ran, in both the solo and prefork pools. It kept running on main, and after a task that succeeds.

CRITICAL/MainProcess] Unrecoverable error: ClientError('An error occurred (InvalidParameterValue) when calling the DeleteMessage operation: Value x for parameter ReceiptHandle is invalid. Reason: The receipt handle has expired.')
  File "kombu/message.py", line 142, in reject_log_error
  File "kombu/message.py", line 164, in reject
  File "kombu/transport/SQS/__init__.py", line 996, in basic_reject
  File "kombu/transport/SQS/__init__.py", line 1010, in _delete_rejected_message

[MehrazRumman]

@ChristianMurphy thanks for testing this so thoroughly. All points are addressed in 15f59d1:

Dead-letter queues: I went with your suggestion. Before deleting, basic_reject reads the queue's RedrivePolicy with GetQueueAttributes. The result is cached per queue URL on the transport, so it is only asked once per queue. If the queue has a redrive policy, the message is left in the queue, so SQS keeps redelivering it until maxReceiveCount moves it to the dead-letter queue. If the call fails, the message is also left in the queue. An AccessDenied answer is cached as well, so a worker without that permission doesn't ask on every reject. The docstring now lists when a rejected message is kept: backoff tasks, queues with a redrive policy, and a failed DeleteMessage.

Failed delete: basic_reject now catches ClientError from DeleteMessage and logs a warning instead of raising. The delivery is still released, so it doesn't hold a prefetch slot, and SQS redelivers the message after its visibility timeout. Unlike basic_ack, it doesn't call basic_reject for other error codes, because releasing the delivery is all that reject has left to do.

Integration tests: I added your test_reject_no_requeue_is_not_redelivered test. I also added test_reject_no_requeue_with_redrive_policy_is_redelivered, which creates a queue with a 1 second visibility timeout and a redrive policy, rejects a message without requeue and checks that it is received again. Against GoAws, the first test fails on main and passes on this branch. The second passes on both. All 42 SQS integration tests pass.

Emergency dump: the reject tests now use a fixture that clears channel.qos._delivered afterwards, so no more EMERGENCY DUMP STATE TO FILE.

[ChristianMurphy]

Looks good. Re-tested 15f59d1 with a Celery worker against moto. With a redrive policy, a failing task runs 3 times and then moves to the dead-letter queue, as on main. Without one, it runs once and the message is deleted. When DeleteMessage fails, the worker logs the error and keeps running, with both the solo and prefork pools. The SQS unit tests no longer write the emergency dump files.

[ishan-1010]

ran 15f59d1. SQS unit tests 316 passed, t/integration/test_sqs.py on moto 40 passed (the 2 test_ack_invalid_receipt_with_backoff cases fail on main under moto too). client selection and the non-ClientError path match basic_ack. one question inline about the redrive cache.

stale cache repro (moto on :4100, from the repo root)

"""A redrive policy added while a worker runs is ignored until reconnect: rejected messages are deleted."""
import json, time, uuid
from unittest.mock import patch
import boto3, kombu, kombu.asynchronous
from t.integration.test_sqs import get_connection

patch("kombu.transport.SQS.SNS._SnsSubscription._set_policy_on_sqs_queue").start()
admin = boto3.client("sqs", region_name="us-east-1", endpoint_url="http://localhost:4100",
                     aws_access_key_id="TestUsername", aws_secret_access_key="TestPassword")
prefix = str(uuid.uuid4())[:8] + "_"
dlq_url = admin.create_queue(QueueName=prefix + "dlq")["QueueUrl"]
dlq_arn = admin.get_queue_attributes(QueueUrl=dlq_url, AttributeNames=["QueueArn"])["Attributes"]["QueueArn"]

def receive_and_reject(conn, q):
    msg = conn.SimpleQueue(q).get(timeout=5)
    msg.reject(requeue=False)
    return msg.payload

conn = get_connection(port=4100, queue_prefix=prefix)
conn.transport_options.update(wait_time_seconds=0, visibility_timeout=1)
q = kombu.Queue("work", routing_key="work")
with conn:
    producer = kombu.Producer(conn.channel())
    producer.publish({"n": 1}, routing_key="work", declare=[q])
    print("reject 1 (no redrive policy yet):", receive_and_reject(conn, q))
    url = conn.default_channel._new_queue("work")
    admin.set_queue_attributes(QueueUrl=url, Attributes={"RedrivePolicy": json.dumps(
        {"deadLetterTargetArn": dlq_arn, "maxReceiveCount": "3"})})
    print("redrive policy added to the live queue")
    producer.publish({"n": 2}, routing_key="work", declare=[q])
    print("reject 2 (same connection):", receive_and_reject(conn, q))
    time.sleep(2)
    left = admin.receive_message(QueueUrl=url, WaitTimeSeconds=1).get("Messages", [])
    print("message 2 still in the queue for SQS to redrive:", bool(left))

conn2 = get_connection(port=4100, queue_prefix=prefix)
conn2.transport_options.update(wait_time_seconds=0, visibility_timeout=1)
with conn2:
    kombu.Producer(conn2.channel()).publish({"n": 3}, routing_key="work", declare=[q])
    receive_and_reject(conn2, q)
    time.sleep(2)
    left = admin.receive_message(QueueUrl=url, WaitTimeSeconds=1).get("Messages", [])
    print("after reconnect, message 3 kept for redrive:", bool(left))

reject 1 (no redrive policy yet): {'n': 1}
redrive policy added to the live queue
reject 2 (same connection): {'n': 2}
message 2 still in the queue for SQS to redrive: False
after reconnect, message 3 kept for redrive: True