Add multiplier parameter to ExponentialBackoff

Co-authored-by: Amp <amp@ampcode.com>

Author: Schnouki

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Added `multiplier` parameter to `ExponentialBackoff` for scaling the exponential term.
+
 ## [0.7.0] - 2025-11-27
 
 ### Added

README.md

@@ -358,6 +358,18 @@ def send_email(to, subject, body):
 
 The default policy is the exponential backoff. We recommend setting both the backoff policy and the dead-letter queue to limit the maximum number of execution attempts.
 
+The `ExponentialBackoff` policy calculates the visibility timeout using the formula:
+
+```
+min_visibility_timeout + multiplier * (base ** attempts)
+```
+
+Parameters:
+- `base` (default: 2): Exponential base in seconds
+- `min_visibility_timeout` (default: 0): Minimum delay in seconds
+- `max_visibility_timeout` (default: 1800): Maximum delay cap in seconds
+- `multiplier` (default: 1): Scaling factor for the exponential term
+
 Alternatively, you can set the backoff to IMMEDIATE_RETURN to re-execute the failed task immediately.
 
 ```python

sqs_workers/backoff_policies.py

@@ -30,14 +30,18 @@ def __init__(
         base: float = 2,
         min_visibility_timeout: float = 0,
         max_visbility_timeout: float = 30 * 60,
+        multiplier: float = 1,
     ) -> None:
         self.base = base  # in seconds
         self.min_visibility_timeout = min_visibility_timeout
         self.max_visibility_timeout = max_visbility_timeout
+        self.multiplier = multiplier
 
     def get_visibility_timeout(self, message) -> int:
         prev_receive_count = int(message.attributes["ApproximateReceiveCount"]) - 1
-        mu = self.min_visibility_timeout + (self.base**prev_receive_count)
+        mu = self.min_visibility_timeout + self.multiplier * (
+            self.base**prev_receive_count
+        )
         sigma = float(mu) / 10
         visibility_timeout = random.normalvariate(mu, sigma)
         visibility_timeout = max(self.min_visibility_timeout, visibility_timeout)

tests/test_backoff_policies.py

@@ -0,0 +1,90 @@
+import random
+from unittest.mock import MagicMock
+
+import pytest
+
+from sqs_workers.backoff_policies import (
+    ConstantBackoff,
+    ExponentialBackoff,
+)
+
+
+@pytest.fixture
+def mock_message():
+    def _make_message(receive_count: int):
+        message = MagicMock()
+        message.attributes = {"ApproximateReceiveCount": str(receive_count)}
+        return message
+
+    return _make_message
+
+
+class TestConstantBackoff:
+    def test_returns_constant_value(self):
+        policy = ConstantBackoff(backoff_value=30)
+        message = MagicMock()
+        assert policy.get_visibility_timeout(message) == 30
+
+    def test_default_is_zero(self):
+        policy = ConstantBackoff()
+        message = MagicMock()
+        assert policy.get_visibility_timeout(message) == 0
+
+
+class TestExponentialBackoff:
+    def test_default_parameters_backward_compatible(self, mock_message):
+        policy = ExponentialBackoff()
+        assert policy.base == 2
+        assert policy.min_visibility_timeout == 0
+        assert policy.max_visibility_timeout == 30 * 60
+        assert policy.multiplier == 1
+
+    def test_first_attempt_uses_base(self, mock_message):
+        random.seed(42)
+        policy = ExponentialBackoff(base=2, min_visibility_timeout=0, multiplier=1)
+        timeout = policy.get_visibility_timeout(mock_message(1))
+        assert timeout == 1
+
+    def test_exponential_growth(self, mock_message):
+        random.seed(42)
+        policy = ExponentialBackoff(base=2, min_visibility_timeout=0, multiplier=1)
+        timeouts = [policy.get_visibility_timeout(mock_message(i)) for i in range(1, 6)]
+        for i in range(1, len(timeouts)):
+            assert timeouts[i] >= timeouts[i - 1]
+
+    def test_multiplier_scales_timeout(self, mock_message):
+        random.seed(42)
+        policy_no_mult = ExponentialBackoff(
+            base=2, min_visibility_timeout=0, multiplier=1
+        )
+        timeout_no_mult = policy_no_mult.get_visibility_timeout(mock_message(3))
+
+        random.seed(42)
+        policy_with_mult = ExponentialBackoff(
+            base=2, min_visibility_timeout=0, multiplier=10
+        )
+        timeout_with_mult = policy_with_mult.get_visibility_timeout(mock_message(3))
+
+        assert timeout_with_mult == pytest.approx(timeout_no_mult * 10, rel=0.2)
+
+    def test_min_visibility_timeout_is_floor(self, mock_message):
+        random.seed(42)
+        policy = ExponentialBackoff(base=2, min_visibility_timeout=60)
+        timeout = policy.get_visibility_timeout(mock_message(1))
+        assert timeout >= 60
+
+    def test_max_visibility_timeout_is_ceiling(self, mock_message):
+        random.seed(42)
+        policy = ExponentialBackoff(
+            base=10, min_visibility_timeout=0, max_visbility_timeout=100, multiplier=100
+        )
+        timeout = policy.get_visibility_timeout(mock_message(10))
+        assert timeout <= 100
+
+    def test_multiplier_combined_with_min_timeout(self, mock_message):
+        random.seed(42)
+        policy = ExponentialBackoff(
+            base=2, min_visibility_timeout=10, multiplier=5, max_visbility_timeout=1000
+        )
+        timeout = policy.get_visibility_timeout(mock_message(3))
+        assert timeout >= 10