Add sts_token_buffer_time parameter to transport options

[eisichenko]

Linked Issue: #2217

During development, I've encountered an issue where Kombu does not currently provide an option to generate an STS token before its expiration time. This can lead to errors such as:

Request HTTP Error HTTP 403 Forbidden (b'{"__type":"com.amazon.coral.service#ExpiredTokenException","message":"The security token included in the request is expired"}')

Such errors occur when the session token expires, which can cause the consumer channel to close when working with Celery.

This PR introduces a new optional parameter sts_token_buffer_time which defaults to 0, preserving the previous behavior. When provided, this attribute allows the STS token to be generated earlier by the specified buffer time seconds (sts_token_buffer_time), helping to prevent ExpiredTokenException errors.

[auvipy]

you need to rebase the PR to fix merge conflict as we got a new PR merged 83b296f . also the pr will need proper unit tests

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 81.61%. Comparing base (13e6938) to head (ab183ac).

Additional details and impacted files

@@           Coverage Diff           @@
##             main    #2216   +/-   ##
=======================================
  Coverage   81.60%   81.61%           
=======================================
  Files          77       77           
  Lines        9540     9545    +5     
  Branches     1162     1163    +1     
=======================================
+ Hits         7785     7790    +5     
  Misses       1563     1563           
  Partials      192      192           

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

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[eisichenko]

you need to rebase the PR to fix merge conflict as we got a new PR merged 83b296f . also the pr will need proper unit tests

@auvipy merge conflicts are resolved and tests are implemented. Thanks for the response!

[auvipy]

I started the CI. So lets see

[eisichenko]

I started the CI. So lets see

@auvipy all checks are passed

[auvipy]

we have to wait for the next release until we merge this, so please have some patience.

[Nusnus]

Lint error:

2025-05-31T16:18:15.2886620Z pydocstyle: commands[0]> pydocstyle /home/runner/_work/kombu/kombu/kombu
2025-05-31T16:18:16.4681757Z /home/runner/_work/kombu/kombu/kombu/transport/SQS.py:787 in public method `generate_sts_session_token_with_buffer`:
2025-05-31T16:18:16.4682415Z         D205: 1 blank line required between summary line and description (found 0)
2025-05-31T16:18:16.4682850Z /home/runner/_work/kombu/kombu/kombu/transport/SQS.py:787 in public method `generate_sts_session_token_with_buffer`:
2025-05-31T16:18:16.4683300Z         D212: Multi-line docstring summary should start at the first line
2025-05-31T16:18:16.4801709Z pydocstyle: exit 1 (1.19 seconds) /home/runner/_work/kombu/kombu> pydocstyle /home/runner/_work/kombu/kombu/kombu pid=4141
2025-05-31T16:18:16.4809844Z   pydocstyle: FAIL code 1 (10.50=setup[9.31]+cmd[1.19] seconds)
2025-05-31T16:18:16.4810170Z   evaluation failed :( (10.59 seconds)
2025-05-31T16:18:16.5340450Z ##[error]Process completed with exit code 1.

[eisichenko]

@Nusnus thanks! Linter issues should be fixed now

[auvipy]

lets consider this for v5.6 as this version will be exclusive for SQS

[auvipy]

@spawn-guy it would be great if you can review this

[Copilot]

Pull Request Overview

This PR adds an optional sts_token_buffer_time parameter to SQS transport options, allowing tokens to be refreshed a specified number of seconds before they expire to avoid ExpiredTokenException errors.

• Introduced sts_token_buffer_time defaulting to 0 alongside existing sts_token_timeout.
• Implemented generate_sts_session_token_with_buffer and wired it into _new_predefined_queue_client_with_sts_session.
• Added documentation and unit tests covering buffer-time behavior.

Reviewed Changes

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

File	Description
t/unit/transport/test_SQS.py	Added tests for new sts_token_buffer_time in both fresh and expired session scenarios.
kombu/transport/SQS.py	Introduced sts_token_buffer_time option, new buffered token generator, docs, and import of timedelta.

Comments suppressed due to low confidence (3)

kombu/transport/SQS.py:807

• Consider validating that token_buffer_seconds is less than token_expiry_seconds and raising a ValueError if not, to avoid silent no-op behavior when an invalid buffer is provided.

    def generate_sts_session_token_with_buffer(self, role_arn, token_expiry_seconds, token_buffer_seconds=0):

kombu/transport/SQS.py:96

• [nitpick] The versionadded description could be sharper. For example: "Buffer in seconds to refresh the STS token before its expiration (default: 0). Must be less than sts_token_timeout."

.. versionadded:: 5.6.0

t/unit/transport/test_SQS.py:1226

• Add a test case where sts_token_buffer_time is equal to or greater than sts_token_timeout to verify that no buffer is applied in that scenario.

    def test_sts_new_session_with_buffer_time(self):

done