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

The request queue exists to preserve heartbeats when the server is
unavailable, but _dispatch_request's error classification defeated it:

- HTTP 503 (sent by aw-server when the heartbeat lock times out, e.g.
  while the server is slow) fell into the 'Unknown error, not retrying'
  branch, permanently discarding the queued request.
- The existing 400/500 branches were dead code: Response.__bool__
  returns Response.ok, which is False for any error status, so
  'if e.response and ...' never matched and every HTTP error was
  dropped - including the 500s the code claimed to retry.
- A plain ConnectionError mid-dispatch (server died after connect) also
  hit the drop branch, and since 'connected' stayed True the dispatch
  loop kept popping and discarding requests one by one - draining the
  entire on-disk queue during a server outage.

Now transient errors (ConnectionError/Timeout, HTTP 429/500/502/503/504)
leave the request at the head of the persistqueue for a later retry,
exactly like ConnectTimeout already did; only permanent client errors
(e.g. HTTP 400 bad payload) are dropped so they can't wedge the queue.
Connection errors also mark the queue disconnected so the run loop goes
back to reconnecting.

Retrying is safe for heartbeats: a duplicate of an already-processed
heartbeat merges into the last event as a no-op, and FIFO head-retry
preserves the ordering that merging requires (dropping requests from the
middle of the stream broke merge chains, fragmenting events).

Author: ErikBjare

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