fix(queue): retry queued requests on transient errors instead of dropping them

aw_client/client.py

@@ -425,6 +425,21 @@ def _warn_queue_before_connect(self) -> None:
 QueuedRequest = namedtuple("QueuedRequest", ["endpoint", "data"])
 Bucket = namedtuple("Bucket", ["id", "type"])
 
+# Bounds for the delay before retrying a queued request after a
+# transient server error (e.g. 429/503), honoring Retry-After if given.
+RETRY_DELAY_DEFAULT = 0.5
+RETRY_DELAY_MAX = 60.0
+
+
+def _retry_delay(response: req.Response) -> float:
+    """Delay before retrying, honoring the Retry-After header (delta-seconds
+    form) if present and sane; the HTTP-date form falls back to the default."""
+    try:
+        delay = float(response.headers.get("Retry-After", RETRY_DELAY_DEFAULT))
+    except ValueError:
+        return RETRY_DELAY_DEFAULT
+    return max(RETRY_DELAY_DEFAULT, min(delay, RETRY_DELAY_MAX))
+
 
 class RequestQueue(threading.Thread):
     """Used to asynchronously send heartbeats.
@@ -436,6 +451,11 @@ class RequestQueue(threading.Thread):
 
     VERSION = 1  # update this whenever the queue-file format changes
 
+    # HTTP statuses that indicate a transient server-side problem, for which
+    # requests are kept in the queue and retried (dropped on anything else).
+    # 503 in particular is sent by aw-server when the heartbeat lock times out.
+    RETRY_STATUS_CODES = {429, 500, 502, 503, 504}
+
     def __init__(self, client: ActivityWatchClient) -> None:
         threading.Thread.__init__(self, daemon=True)
 
@@ -515,13 +535,12 @@ def _dispatch_request(self) -> None:
 
         try:
             self.client._post(request.endpoint, request.data)
-        except req.exceptions.ConnectTimeout:
+        except (req.exceptions.ConnectionError, req.exceptions.Timeout):
             # Triggered by:
             #   - server not running (connection refused)
             #   - server not responding (timeout)
-            # Safe to retry according to requests docs:
-            #   https://requests.readthedocs.io/en/latest/api/#requests.ConnectTimeout
-
+            # Keep the request in the queue and go back to waiting for the
+            # server to become available (the run loop reconnects).
             self.connected = False
             logger.warning(
                 "Connection refused or timeout, will queue requests until connection is available."
@@ -532,20 +551,32 @@ def _dispatch_request(self) -> None:
             sleep(0.5)
             return
         except req.RequestException as e:
-            if e.response and e.response.status_code == 400:
-                # HTTP 400 - Bad request
-                # Example case: https://github.com/ActivityWatch/activitywatch/issues/815
-                # We don't want to retry, because a bad payload is likely to fail forever.
-                logger.error(f"Bad request, not retrying: {request.data}")
-            elif e.response and e.response.status_code == 500:
-                # HTTP 500 - Internal server error
-                # It is possible that the server is in a bad state (and will recover on restart),
-                # in which case we want to retry. I hope this can never caused by a bad payload.
-                logger.error(f"Internal server error, retrying: {request.data}")
-                sleep(0.5)
+            # NOTE: `e.response is not None` matters: Response.__bool__ is
+            # False for any non-2xx status, so a plain `if e.response` never
+            # matches an error response.
+            response = e.response
+            status_code = response.status_code if response is not None else None
+            if response is not None and status_code in self.RETRY_STATUS_CODES:
+                # Transient server-side problem (busy, overloaded, restarting
+                # or behind a flaky proxy) - the request itself is likely
+                # fine, so keep it in the queue and retry. Heartbeats are safe
+                # to replay: a duplicate of an already-processed heartbeat
+                # merges into the last event as a no-op.
+                delay = _retry_delay(response)
+                logger.warning(
+                    f"Server error {status_code}, will retry in {delay}s: {request.endpoint}"
+                )
+                # stop-aware wait, so a long Retry-After can't block shutdown
+                self.wait(delay)
                 return
             else:
-                logger.exception(f"Unknown error, not retrying: {request.data}")
+                # Client errors (e.g. HTTP 400 - bad request, see
+                # https://github.com/ActivityWatch/activitywatch/issues/815)
+                # are likely to fail forever, so drop the request instead of
+                # blocking the queue.
+                logger.error(
+                    f"Request failed ({status_code}), not retrying: {request.data}"
+                )
         except Exception:
             logger.exception(f"Unknown error, not retrying: {request.data}")
 

tests/test_requestqueue.py

@@ -12,7 +12,9 @@
 
 basicConfig(level=DEBUG)
 
+import pytest
 import requests
+
 from aw_client.client import RequestQueue
 
 
@@ -88,3 +90,108 @@ def create_bucket(self, *args, **kwargs):
 
     assert rq.connected is False
     assert client.create_bucket_calls == [(("test-bucket", "test-type"), {})]
+
+
+def _http_error(status_code: int) -> requests.exceptions.HTTPError:
+    response = requests.Response()
+    response.status_code = status_code
+    return requests.exceptions.HTTPError(response=response)
+
+
+class FlakyClient(MockClient):
+    """Client whose _post raises the given exception until cleared."""
+
+    def __init__(self, exc):
+        super().__init__()
+        self.exc = exc
+        self.post_calls = 0
+
+    def _post(self, *args, **kwargs):
+        self.post_calls += 1
+        if self.exc:
+            raise self.exc
+        return requests.Response()
+
+
+def _fresh_queue(client) -> RequestQueue:
+    """Create a RequestQueue and drain requests persisted by earlier runs."""
+    rq = RequestQueue(client)  # type: ignore
+    while rq._get_next():
+        rq._task_done()
+    return rq
+
+
+@pytest.mark.parametrize("status_code", [429, 500, 502, 503, 504])
+def test_dispatch_retries_transient_server_errors(status_code):
+    """
+    Transient server-side errors (e.g. 503 from aw-server's heartbeat-lock
+    timeout) must keep the request in the queue for a later retry, then
+    dispatch it once the server recovers.
+
+    Also guards against the Response.__bool__ pitfall: `if e.response` is
+    False for any error status, which used to send every HTTP error down the
+    "not retrying" path (dropping the request permanently).
+    """
+    client = FlakyClient(_http_error(status_code))
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    rq.add_request("buckets/test/heartbeat?pulsetime=10", {"label": "test"})
+    rq._dispatch_request()
+
+    assert client.post_calls == 1
+    assert rq._get_next() is not None  # still queued
+
+    client.exc = None  # server recovered
+    rq._dispatch_request()
+
+    assert client.post_calls == 2
+    assert rq._get_next() is None  # delivered and popped
+
+
+def test_dispatch_drops_client_errors():
+    """A bad payload (HTTP 400) fails forever and must not block the queue."""
+    client = FlakyClient(_http_error(400))
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    rq.add_request("buckets/test/heartbeat?pulsetime=10", {"label": "bad"})
+    rq._dispatch_request()
+
+    assert client.post_calls == 1
+    assert rq._get_next() is None  # dropped
+
+
+def test_dispatch_keeps_queue_on_connection_error():
+    """
+    A connection error mid-dispatch (server died after connect) must keep the
+    request queued and mark the queue disconnected, so the run loop goes back
+    to reconnecting instead of draining the queue into the void.
+    """
+    client = FlakyClient(requests.exceptions.ConnectionError())
+    rq = _fresh_queue(client)
+    rq.connected = True
+
+    rq.add_request("buckets/test/heartbeat?pulsetime=10", {"label": "test"})
+    rq._dispatch_request()
+
+    assert rq._get_next() is not None  # still queued
+    assert rq.connected is False
+
+
+def test_retry_delay_honors_retry_after():
+    from aw_client.client import _retry_delay, RETRY_DELAY_DEFAULT, RETRY_DELAY_MAX
+
+    def resp(retry_after=None):
+        r = requests.Response()
+        r.status_code = 429
+        if retry_after is not None:
+            r.headers["Retry-After"] = retry_after
+        return r
+
+    assert _retry_delay(resp()) == RETRY_DELAY_DEFAULT  # absent
+    assert _retry_delay(resp("2")) == 2.0  # delta-seconds
+    assert _retry_delay(resp("9999")) == RETRY_DELAY_MAX  # capped
+    assert _retry_delay(resp("0")) == RETRY_DELAY_DEFAULT  # floored
+    # HTTP-date form is not parsed, falls back to default
+    assert _retry_delay(resp("Wed, 21 Oct 2026 07:28:00 GMT")) == RETRY_DELAY_DEFAULT