diff --git a/aw_client/client.py b/aw_client/client.py index f176982..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. @@ -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}") diff --git a/tests/test_requestqueue.py b/tests/test_requestqueue.py index 836b69c..b6d2dab 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,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