[DrewRubin]

Issue

Hi, this is a small change that address two issues:

1. There is no cancellation of previous timers for the curl timeout, so if the SQS server becomes unavailable, it schedules a large number of them.
2. The units are mismatched, which worsens the issue as they are scheduled far in the future.

I reproduced this locally by stopping a running ElasticMQ server and then saw memory growth and high CPU usage.

self.hub.call_later(msecs, self._timeout_check)

Leads to

def call_later(self, delay, callback, *args):
    return self.timer.call_after(delay, callback, args)

Which calls call_after in seconds

def call_after(self, secs, fun, args=(), kwargs=None, priority=0):
    kwargs = {} if not kwargs else kwargs
    return self.enter_after(secs, self.Entry(fun, args, kwargs), priority)

Fix

We should probably only keep the latest timer, so use return value instead of creating new, and cancels previous if it exists. Did unit conversion in-function to maintain contract.

[auvipy]

we need tests to verify this change

[DrewRubin]

@auvipy Added a test for the subsequent timer cancelation

[auvipy]

any relevant old issues or discussions for this fix?

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 85.13%. Comparing base (edf7c0a) to head (3206937).

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2567   +/-   ##
=======================================
  Coverage   85.12%   85.13%           
=======================================
  Files          80       80           
  Lines       11371    11376    +5     
  Branches     1363     1365    +2     
=======================================
+ Hits         9680     9685    +5     
  Misses       1471     1471           
  Partials      220      220           

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

[DrewRubin]

@auvipy Not that I saw, it's just something I noticed and monkey patched in our system, so figured I would try to fix upstream.

[Copilot]

Pull request overview

This pull request fixes curl multi timeout scheduling in CurlClient by converting pycurl’s millisecond timeout value into the hub’s expected seconds and by cancelling the previously scheduled timeout to prevent timer buildup (and related memory/CPU growth) when an endpoint becomes unavailable.

Changes:

• Convert _set_timeout input from milliseconds to seconds before calling hub.call_later.
• Store the scheduled timer reference and cancel the previous one before scheduling a new timeout.
• Update/add unit tests to validate the new timeout unit and the timer-cancellation behavior.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.

File	Description
kombu/asynchronous/http/curl.py	Converts timeout units and tracks/cancels the outstanding hub timer to prevent accumulating scheduled timeouts.
t/unit/asynchronous/http/test_curl.py	Updates the unit expectation for _set_timeout and adds coverage for cancelling the previous scheduled timer.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[auvipy]

can you please cross check this pr #2488 and see if we can incorporate the approached followed there?

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

Suppressed comments (1)

kombu/asynchronous/http/curl.py:61

• The newly tracked one-shot timer is not cancelled by close(). If the client closes while a curl timeout is pending, the hub retains the client until that deadline and then invokes _timeout_check on the already-closed CurlMulti; cancel _tref alongside the existing repeated timer.

        self._tref = None

[XDanielPaul]

First of all, sorry folks if this is unrelated, I came here from #2488.

can you please cross check this pr #2488 and see if we can incorporate the approached followed there?

+1 for this

I've encountered an issue based on this: If a network interruption silently drops the response to an in-flight ReceiveMessage long-poll (the TCP connection stays open, but no data, FIN or RST ever arrives), the Celery worker stops consuming from SQS permanently. The worker process stays up and healthy, logs no errors, and messages pile up in the queue until the worker is restarted.

My investigation found out that kombu/asynchronous/aws/connection.py builds every request with connect_timeout and request_timeout (20 s by default in AsyncHTTPSConnection). But CurlClient._setup_request in kombu/asynchronous/http/curl.py never sets pycurl.TIMEOUT/TIMEOUT_MS or CONNECTTIMEOUT/CONNECTTIMEOUT_MS on the curl handle, so those values are ignored. If the response never arrives, curl waits indefinitely, the request never completes, and the next poll is never scheduled.

I am aware that this is probably an edge case, but it happened in a production environment.

[auvipy]

thanks for the ping