feat(api_core): make the last retry happen at deadline (#9873)

* feat(api_core): allow setting Retry deadline as strict

If a deadline is set as strict, Retry will shorten the last sleep
period to end at the given deadline, and not possibly stretch beyond
it.

* feat(core): make the last retry happen at deadline

This commit changes Retry instances in a way that the last sleep period
is shortened so that its end matches at the given deadline. This
prevents the last sleep period to last way beyond the deadline.

api_core/google/api_core/retry.py

@@ -155,10 +155,12 @@ def retry_target(target, predicate, sleep_generator, deadline, on_error=None):
             It should return True to retry or False otherwise.
         sleep_generator (Iterable[float]): An infinite iterator that determines
             how long to sleep between retries.
-        deadline (float): How long to keep retrying the target.
-        on_error (Callable): A function to call while processing a retryable
-            exception.  Any error raised by this function will *not* be
-            caught.
+        deadline (float): How long to keep retrying the target. The last sleep
+            period is shortened as necessary, so that the last retry runs at
+            ``deadline`` (and not considerably beyond it).
+        on_error (Callable[Exception]): A function to call while processing a
+            retryable exception.  Any error raised by this function will *not*
+            be caught.
 
     Returns:
         Any: the return value of the target function.
@@ -191,16 +193,21 @@ def retry_target(target, predicate, sleep_generator, deadline, on_error=None):
                 on_error(exc)
 
         now = datetime_helpers.utcnow()
-        if deadline_datetime is not None and deadline_datetime < now:
-            six.raise_from(
-                exceptions.RetryError(
-                    "Deadline of {:.1f}s exceeded while calling {}".format(
-                        deadline, target
+
+        if deadline_datetime is not None:
+            if deadline_datetime <= now:
+                six.raise_from(
+                    exceptions.RetryError(
+                        "Deadline of {:.1f}s exceeded while calling {}".format(
+                            deadline, target
+                        ),
+                        last_exc,
                     ),
                     last_exc,
-                ),
-                last_exc,
-            )
+                )
+            else:
+                time_to_deadline = (deadline_datetime - now).total_seconds()
+                sleep = min(time_to_deadline, sleep)
 
         _LOGGER.debug(
             "Retrying due to {}, sleeping {:.1f}s ...".format(last_exc, sleep)
@@ -227,7 +234,9 @@ class Retry(object):
             must be greater than 0.
         maximum (float): The maximum amout of time to delay in seconds.
         multiplier (float): The multiplier applied to the delay.
-        deadline (float): How long to keep retrying in seconds.
+        deadline (float): How long to keep retrying in seconds. The last sleep
+            period is shortened as necessary, so that the last retry runs at
+            ``deadline`` (and not considerably beyond it).
     """
 
     def __init__(
@@ -251,8 +260,8 @@ def __call__(self, func, on_error=None):
 
         Args:
             func (Callable): The callable to add retry behavior to.
-            on_error (Callable): A function to call while processing a
-                retryable exception.  Any error raised by this function will
+            on_error (Callable[Exception]): A function to call while processing
+                a retryable exception. Any error raised by this function will
                 *not* be caught.
 
         Returns:

api_core/tests/unit/test_retry.py

@@ -182,35 +182,94 @@ def test_constructor_options(self):
         assert retry_._on_error is _some_function
 
     def test_with_deadline(self):
-        retry_ = retry.Retry()
+        retry_ = retry.Retry(
+            predicate=mock.sentinel.predicate,
+            initial=1,
+            maximum=2,
+            multiplier=3,
+            deadline=4,
+            on_error=mock.sentinel.on_error,
+        )
         new_retry = retry_.with_deadline(42)
         assert retry_ is not new_retry
         assert new_retry._deadline == 42
 
+        # the rest of the attributes should remain the same
+        assert new_retry._predicate is retry_._predicate
+        assert new_retry._initial == retry_._initial
+        assert new_retry._maximum == retry_._maximum
+        assert new_retry._multiplier == retry_._multiplier
+        assert new_retry._on_error is retry_._on_error
+
     def test_with_predicate(self):
-        retry_ = retry.Retry()
+        retry_ = retry.Retry(
+            predicate=mock.sentinel.predicate,
+            initial=1,
+            maximum=2,
+            multiplier=3,
+            deadline=4,
+            on_error=mock.sentinel.on_error,
+        )
         new_retry = retry_.with_predicate(mock.sentinel.predicate)
         assert retry_ is not new_retry
         assert new_retry._predicate == mock.sentinel.predicate
 
+        # the rest of the attributes should remain the same
+        assert new_retry._deadline == retry_._deadline
+        assert new_retry._initial == retry_._initial
+        assert new_retry._maximum == retry_._maximum
+        assert new_retry._multiplier == retry_._multiplier
+        assert new_retry._on_error is retry_._on_error
+
     def test_with_delay_noop(self):
-        retry_ = retry.Retry()
+        retry_ = retry.Retry(
+            predicate=mock.sentinel.predicate,
+            initial=1,
+            maximum=2,
+            multiplier=3,
+            deadline=4,
+            on_error=mock.sentinel.on_error,
+        )
         new_retry = retry_.with_delay()
         assert retry_ is not new_retry
         assert new_retry._initial == retry_._initial
         assert new_retry._maximum == retry_._maximum
         assert new_retry._multiplier == retry_._multiplier
 
     def test_with_delay(self):
-        retry_ = retry.Retry()
+        retry_ = retry.Retry(
+            predicate=mock.sentinel.predicate,
+            initial=1,
+            maximum=2,
+            multiplier=3,
+            deadline=4,
+            on_error=mock.sentinel.on_error,
+        )
         new_retry = retry_.with_delay(initial=1, maximum=2, multiplier=3)
         assert retry_ is not new_retry
         assert new_retry._initial == 1
         assert new_retry._maximum == 2
         assert new_retry._multiplier == 3
 
+        # the rest of the attributes should remain the same
+        assert new_retry._deadline == retry_._deadline
+        assert new_retry._predicate is retry_._predicate
+        assert new_retry._on_error is retry_._on_error
+
     def test___str__(self):
-        retry_ = retry.Retry()
+        def if_exception_type(exc):
+            return bool(exc)  # pragma: NO COVER
+
+        # Explicitly set all attributes as changed Retry defaults should not
+        # cause this test to start failing.
+        retry_ = retry.Retry(
+            predicate=if_exception_type,
+            initial=1.0,
+            maximum=60.0,
+            multiplier=2.0,
+            deadline=120.0,
+            on_error=None,
+        )
         assert re.match(
             (
                 r"<Retry predicate=<function.*?if_exception_type.*?>, "
@@ -259,6 +318,54 @@ def test___call___and_execute_retry(self, sleep, uniform):
         sleep.assert_called_once_with(retry_._initial)
         assert on_error.call_count == 1
 
+    # Make uniform return half of its maximum, which is the calculated sleep time.
+    @mock.patch("random.uniform", autospec=True, side_effect=lambda m, n: n / 2.0)
+    @mock.patch("time.sleep", autospec=True)
+    def test___call___and_execute_retry_hitting_deadline(self, sleep, uniform):
+
+        on_error = mock.Mock(spec=["__call__"], side_effect=[None] * 10)
+        retry_ = retry.Retry(
+            predicate=retry.if_exception_type(ValueError),
+            initial=1.0,
+            maximum=1024.0,
+            multiplier=2.0,
+            deadline=9.9,
+        )
+
+        utcnow = datetime.datetime.utcnow()
+        utcnow_patcher = mock.patch(
+            "google.api_core.datetime_helpers.utcnow", return_value=utcnow
+        )
+
+        target = mock.Mock(spec=["__call__"], side_effect=[ValueError()] * 10)
+        # __name__ is needed by functools.partial.
+        target.__name__ = "target"
+
+        decorated = retry_(target, on_error=on_error)
+        target.assert_not_called()
+
+        with utcnow_patcher as patched_utcnow:
+            # Make sure that calls to fake time.sleep() also advance the mocked
+            # time clock.
+            def increase_time(sleep_delay):
+                patched_utcnow.return_value += datetime.timedelta(seconds=sleep_delay)
+            sleep.side_effect = increase_time
+
+            with pytest.raises(exceptions.RetryError):
+                decorated("meep")
+
+        assert target.call_count == 5
+        target.assert_has_calls([mock.call("meep")] * 5)
+        assert on_error.call_count == 5
+
+        # check the delays
+        assert sleep.call_count == 4  # once between each successive target calls
+        last_wait = sleep.call_args.args[0]
+        total_wait = sum(call_args.args[0] for call_args in sleep.call_args_list)
+
+        assert last_wait == 2.9   # and not 8.0, because the last delay was shortened
+        assert total_wait == 9.9  # the same as the deadline
+
     @mock.patch("time.sleep", autospec=True)
     def test___init___without_retry_executed(self, sleep):
         _some_function = mock.Mock()