queues: race can cause RetryAfter timeout to be overridden by default

[msullivan]

test_push_callback_retries_until_worker_channel_is_available flaked in a real CI run.

I think there are actually two possibilities here, one of which is documented:

1. Stopping the retry task times out
2. The request for a lease extension with the default timeout has already been made to the server when the cancel happens, and it gets delayed in the server for whatever reason

Here's a diff where I inserted sleeps to force it to come up.

diff --git a/integrations/vercel-celery/tests/integration/test_celery_vqs_worker.py b/integrations/vercel-celery/tests/integration/test_celery_vqs_worker.py
index 8c54845..c839bcd 100644
--- a/integrations/vercel-celery/tests/integration/test_celery_vqs_worker.py
+++ b/integrations/vercel-celery/tests/integration/test_celery_vqs_worker.py
@@ -140,6 +140,7 @@ def test_push_callback_retries_until_worker_channel_is_available(
     push_worker: WorkerApp,
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
+    print()
     monkeypatch.setattr(vqs_celery, "_PUSH_CHANNEL_WAIT_SECONDS", 0.0)
     result = push_worker.add.apply_async((8, 13), queue="emails")
     push_channels = list(vqs_celery._push_channels)
@@ -151,6 +152,10 @@ def test_push_callback_retries_until_worker_channel_is_available(
 
     message = isolated_eqs.state.messages[0]
     assert message.acknowledged is False
+
+    time.sleep(0.3)
+    print("CHECK", time.monotonic())
+
     assert message.lease_deadline_by_consumer["celery"] == (
         isolated_eqs.state.now + timedelta(seconds=1)
     )
diff --git a/src/vercel-queue/vercel/queue/_internal/client_sync.py b/src/vercel-queue/vercel/queue/_internal/client_sync.py
index bb74737..ccc3ba6 100644
--- a/src/vercel-queue/vercel/queue/_internal/client_sync.py
+++ b/src/vercel-queue/vercel/queue/_internal/client_sync.py
@@ -405,6 +405,9 @@ class _MessageLifecycle:
         # directive: the worst remaining case is the pre-existing race, while
         # refusing to apply RetryAfter would always leave the long lease behind.
         wait_for_renewal_stop = isinstance(exc, (Handoff, RetryAfter))
+        if wait_for_renewal_stop:
+            import time
+            time.sleep(1)
         try:
             self._renewal.stop(wait=wait_for_renewal_stop)
         finally:
diff --git a/src/vercel-queue/vercel/queue/_internal/embedded.py b/src/vercel-queue/vercel/queue/_internal/embedded.py
index 5b678da..fb43afe 100644
--- a/src/vercel-queue/vercel/queue/_internal/embedded.py
+++ b/src/vercel-queue/vercel/queue/_internal/embedded.py
@@ -1121,6 +1121,13 @@ class EmbeddedQueueAsgiApp:
             seconds = _visibility_timeout_seconds_from_body(body)
         except (TypeError, ValueError, json.JSONDecodeError) as exc:
             return _bad_request(str(exc))
+
+        import asyncio
+        print("PATCH", seconds, time.monotonic())
+        if seconds == 30:
+            await asyncio.sleep(1.1)
+            print(".....", time.monotonic())
+
         new_deadline = self._server.now + timedelta(seconds=seconds)
         if new_deadline > receipt_message.expires_at:
             return _json_response(