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

aw_client/client.py

@@ -436,6 +436,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 +520,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 +536,29 @@ 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}")
+            # 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.
+            status_code = e.response.status_code if e.response is not None else None
+            if 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.
+                logger.warning(
+                    f"Server error {status_code}, will retry: {request.endpoint}"
+                )
                 sleep(0.5)
                 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,90 @@ 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