From b2f29281f87bf175a6b77b472f9b65bee1e0a7f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Erik=20Bj=C3=A4reholt?= Date: Thu, 23 Jul 2026 12:59:44 +0200 Subject: [PATCH 1/2] 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). --- aw_client/client.py | 43 +++++++++++------- tests/test_requestqueue.py | 89 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 117 insertions(+), 15 deletions(-) diff --git a/aw_client/client.py b/aw_client/client.py index f176982..d8c42e0 100644 --- a/aw_client/client.py +++ b/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}") diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py index 836b69c..6dd1838 100644 --- a/tests/test_requestqueue.py +++ b/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 From b31ba5a7526ed641047f2630aa433c394b342f8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Erik=20Bj=C3=A4reholt?= Date: Thu, 23 Jul 2026 13:06:49 +0200 Subject: [PATCH 2/2] fix(queue): honor Retry-After on transient errors (review feedback) Use the Retry-After header (delta-seconds form, floored at 0.5s and capped at 60s) for the retry delay on 429/503-style responses, and use the stop-aware wait() instead of sleep() so a long delay can't block shutdown. --- aw_client/client.py | 26 ++++++++++++++++++++++---- tests/test_requestqueue.py | 18 ++++++++++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) diff --git a/aw_client/client.py b/aw_client/client.py index d8c42e0..d0682c0 100644 --- a/aw_client/client.py +++ b/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. @@ -539,17 +554,20 @@ def _dispatch_request(self) -> None: # 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: + 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: {request.endpoint}" + f"Server error {status_code}, will retry in {delay}s: {request.endpoint}" ) - sleep(0.5) + # stop-aware wait, so a long Retry-After can't block shutdown + self.wait(delay) return else: # Client errors (e.g. HTTP 400 - bad request, see diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py index 6dd1838..b6d2dab 100644 --- a/tests/test_requestqueue.py +++ b/tests/test_requestqueue.py @@ -177,3 +177,21 @@ def test_dispatch_keeps_queue_on_connection_error(): 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