Skip to content

Commit a19a73f

Browse files
Bhanu-Vardhan-ReddyBhanu-Vardhan-Reddy
committed
Implement retry logic for payload sending in client and server
- Added a retry mechanism for sending payloads in both client and server modules, with configurable attempts and delay. - Refactored `_send_payload_async` function to handle serialization and websocket communication with retries on failure. - Updated `_fetch_proj_auth_cached` to include retry logic for fetching project authentication. - Enhanced error handling and logging during retry attempts to improve user feedback and robustness.
1 parent 11953e1 commit a19a73f

3 files changed

Lines changed: 149 additions & 53 deletions

File tree

‎auralogger/client/client_log.py‎

Lines changed: 67 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
UNKNOWN_TYPE = "unknown"
2323
CONNECT_TIMEOUT_S = 5
2424
DEFAULT_SOCKET_IDLE_CLOSE_MS = 60_000
25+
SDK_RETRY_ATTEMPTS = 3
26+
SDK_RETRY_DELAY_S = 0.5
2527

2628
_ws: Optional[Any] = None
2729
_bound_url: Optional[str] = None
@@ -89,9 +91,20 @@ def _fetch_proj_auth_cached(project_token: str) -> Optional[Dict[str, Any]]:
8991
with _hydrate_lock:
9092
if _hydration_cache_token == project_token and _hydration_cache_raw is not None:
9193
return _hydration_cache_raw
92-
try:
93-
raw = fetch_proj_auth_payload(project_token)
94-
except ValueError:
94+
raw = None
95+
for attempt in range(1, SDK_RETRY_ATTEMPTS + 1):
96+
try:
97+
raw = fetch_proj_auth_payload(project_token)
98+
break
99+
except ValueError:
100+
if attempt >= SDK_RETRY_ATTEMPTS:
101+
return None
102+
print(
103+
f"auralogger: proj_auth failed; retrying ({attempt + 1}/{SDK_RETRY_ATTEMPTS})...",
104+
file=sys.stderr,
105+
)
106+
threading.Event().wait(SDK_RETRY_DELAY_S)
107+
if raw is None:
95108
return None
96109
_hydration_cache_token = project_token
97110
_hydration_cache_raw = raw
@@ -212,6 +225,41 @@ def _styles_for_console(
212225
return None
213226

214227

228+
def _send_payload_async(project_token: str, payload: Dict[str, Any]) -> None:
229+
try:
230+
body = json.dumps(payload)
231+
except (TypeError, ValueError) as e:
232+
print(f"auralogger: failed to serialize log payload: {e}", file=sys.stderr)
233+
return
234+
235+
for attempt in range(1, SDK_RETRY_ATTEMPTS + 1):
236+
try:
237+
ws = _ensure_ws(project_token)
238+
ws.send(body)
239+
_schedule_socket_idle_close()
240+
return
241+
except websocket.WebSocketTimeoutException as e:
242+
close_client_log_socket()
243+
if attempt >= SDK_RETRY_ATTEMPTS:
244+
print(f"auralogger: websocket send failed (timeout): {e}", file=sys.stderr)
245+
return
246+
print(
247+
f"auralogger: websocket send timeout; retrying ({attempt + 1}/{SDK_RETRY_ATTEMPTS})...",
248+
file=sys.stderr,
249+
)
250+
threading.Event().wait(SDK_RETRY_DELAY_S)
251+
except Exception as e:
252+
close_client_log_socket()
253+
if attempt >= SDK_RETRY_ATTEMPTS:
254+
print(f"auralogger: websocket send failed: {e}", file=sys.stderr)
255+
return
256+
print(
257+
f"auralogger: websocket send failed ({e}); retrying ({attempt + 1}/{SDK_RETRY_ATTEMPTS})...",
258+
file=sys.stderr,
259+
)
260+
threading.Event().wait(SDK_RETRY_DELAY_S)
261+
262+
215263
def client_log(
216264
type: str,
217265
message: str,
@@ -221,19 +269,16 @@ def client_log(
221269
global _warned_missing_project_token, _warned_missing_project_id
222270

223271
project_token = _resolve_project_token_runtime()
224-
if not project_token:
225-
if not _warned_missing_project_token:
226-
_warned_missing_project_token = True
227-
print(
228-
"auralogger: missing project token. Call AuraClient.configure(project_token) "
229-
"or set AURALOGGER_PROJECT_TOKEN.",
230-
file=sys.stderr,
231-
)
232-
return
233-
234-
merged = _merged_runtime_for_send(project_token)
235-
can_send = merged is not None
236-
if not can_send and not _warned_missing_project_id:
272+
merged = _merged_runtime_for_send(project_token) if project_token else None
273+
can_send = bool(project_token and merged is not None)
274+
if not project_token and not _warned_missing_project_token:
275+
_warned_missing_project_token = True
276+
print(
277+
"auralogger: missing project token. Call AuraClient.configure(project_token) "
278+
"or set AURALOGGER_PROJECT_TOKEN.",
279+
file=sys.stderr,
280+
)
281+
if project_token and not can_send and not _warned_missing_project_id:
237282
_warned_missing_project_id = True
238283
print(
239284
"auralogger: client logger is running in console-only mode. "
@@ -267,22 +312,12 @@ def client_log(
267312
if not can_send:
268313
return
269314

270-
try:
271-
body = json.dumps(payload)
272-
except (TypeError, ValueError) as e:
273-
print(f"auralogger: failed to serialize log payload: {e}", file=sys.stderr)
274-
return
275-
276-
try:
277-
ws = _ensure_ws(project_token)
278-
ws.send(body)
279-
_schedule_socket_idle_close()
280-
except websocket.WebSocketTimeoutException as e:
281-
print(f"auralogger: websocket send failed (timeout): {e}", file=sys.stderr)
282-
close_client_log_socket()
283-
except Exception as e:
284-
print(f"auralogger: websocket send failed: {e}", file=sys.stderr)
285-
close_client_log_socket()
315+
assert project_token is not None
316+
threading.Thread(
317+
target=_send_payload_async,
318+
args=(project_token, payload.copy()),
319+
daemon=True,
320+
).start()
286321

287322

288323
def auralog(loginputs: ClientLogInputs) -> None:

‎auralogger/server/aura_log.py‎

Lines changed: 57 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525

2626
UNKNOWN_TYPE = "unknown"
2727
CONNECT_TIMEOUT_S = 5
28+
SDK_RETRY_ATTEMPTS = 3
29+
SDK_RETRY_DELAY_S = 0.5
2830

2931
_ws: Optional[Any] = None
3032
_bound_url: Optional[str] = None
@@ -118,9 +120,20 @@ def _fetch_proj_auth_cached(project_token: str) -> Optional[Dict[str, Any]]:
118120
with _hydrate_lock:
119121
if _hydration_cache_token == project_token and _hydration_cache_raw is not None:
120122
return _hydration_cache_raw
121-
try:
122-
raw = fetch_proj_auth_payload(project_token)
123-
except ValueError:
123+
raw = None
124+
for attempt in range(1, SDK_RETRY_ATTEMPTS + 1):
125+
try:
126+
raw = fetch_proj_auth_payload(project_token)
127+
break
128+
except ValueError:
129+
if attempt >= SDK_RETRY_ATTEMPTS:
130+
return None
131+
print(
132+
f"auralogger: proj_auth failed; retrying ({attempt + 1}/{SDK_RETRY_ATTEMPTS})...",
133+
file=sys.stderr,
134+
)
135+
threading.Event().wait(SDK_RETRY_DELAY_S)
136+
if raw is None:
124137
return None
125138
_hydration_cache_token = project_token
126139
_hydration_cache_raw = raw
@@ -212,6 +225,40 @@ def _styles_for_console(
212225
return None
213226

214227

228+
def _send_payload_async(project_token: str, user_secret: str, payload: Dict[str, Any]) -> None:
229+
try:
230+
body = json.dumps(payload)
231+
except (TypeError, ValueError) as e:
232+
print(f"auralogger: failed to serialize log payload: {e}", file=sys.stderr)
233+
return
234+
235+
for attempt in range(1, SDK_RETRY_ATTEMPTS + 1):
236+
try:
237+
ws = _ensure_ws(project_token, user_secret)
238+
ws.send(body)
239+
return
240+
except websocket.WebSocketTimeoutException as e:
241+
close_aura_log_socket()
242+
if attempt >= SDK_RETRY_ATTEMPTS:
243+
print(f"auralogger: websocket send failed (timeout): {e}", file=sys.stderr)
244+
return
245+
print(
246+
f"auralogger: websocket send timeout; retrying ({attempt + 1}/{SDK_RETRY_ATTEMPTS})...",
247+
file=sys.stderr,
248+
)
249+
threading.Event().wait(SDK_RETRY_DELAY_S)
250+
except Exception as e:
251+
close_aura_log_socket()
252+
if attempt >= SDK_RETRY_ATTEMPTS:
253+
print(f"auralogger: websocket send failed: {e}", file=sys.stderr)
254+
return
255+
print(
256+
f"auralogger: websocket send failed ({e}); retrying ({attempt + 1}/{SDK_RETRY_ATTEMPTS})...",
257+
file=sys.stderr,
258+
)
259+
threading.Event().wait(SDK_RETRY_DELAY_S)
260+
261+
215262
def aura_log(
216263
type: str,
217264
message: str,
@@ -279,24 +326,13 @@ def aura_log(
279326
assert project_token is not None
280327
assert user_secret is not None
281328

282-
remote_session = merged["session"]
283-
payload["session"] = remote_session
284-
285-
try:
286-
body = json.dumps(payload)
287-
except (TypeError, ValueError) as e:
288-
print(f"auralogger: failed to serialize log payload: {e}", file=sys.stderr)
289-
return
290-
291-
try:
292-
ws = _ensure_ws(project_token, user_secret)
293-
ws.send(body)
294-
except websocket.WebSocketTimeoutException as e:
295-
print(f"auralogger: websocket send failed (timeout): {e}", file=sys.stderr)
296-
close_aura_log_socket()
297-
except Exception as e:
298-
print(f"auralogger: websocket send failed: {e}", file=sys.stderr)
299-
close_aura_log_socket()
329+
send_payload = payload.copy()
330+
send_payload["session"] = merged["session"]
331+
threading.Thread(
332+
target=_send_payload_async,
333+
args=(project_token, user_secret, send_payload),
334+
daemon=True,
335+
).start()
300336

301337

302338
class AuraServer:

‎tests/test_client_sdk.py‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,16 @@ def close(self) -> None:
2424
self.closed = True
2525

2626

27+
class _ImmediateThread:
28+
def __init__(self, target, args=(), daemon=None) -> None:
29+
self._target = target
30+
self._args = args
31+
self.daemon = daemon
32+
33+
def start(self) -> None:
34+
self._target(*self._args)
35+
36+
2737
class AuraClientSdkTests(unittest.TestCase):
2838
def test_configure_rejects_empty_token(self) -> None:
2939
with self.assertRaises(ValueError):
@@ -66,12 +76,27 @@ def test_client_log_sends_when_runtime_is_hydrated(self) -> None:
6676
"auralogger.client.client_log._ensure_ws",
6777
return_value=ws,
6878
), patch("auralogger.client.client_log.print_log"), patch(
79+
"auralogger.client.client_log.threading.Thread",
80+
_ImmediateThread,
81+
), patch(
6982
"auralogger.client.client_log._schedule_socket_idle_close"
7083
):
7184
client_log("info", "hello", "tests/client", {"x": 1})
7285

7386
self.assertEqual(len(ws.sent), 1)
7487

88+
def test_client_log_prints_even_without_project_token(self) -> None:
89+
with patch(
90+
"auralogger.client.client_log._resolve_project_token_runtime",
91+
return_value=None,
92+
), patch("auralogger.client.client_log.print_log") as mocked_print, patch(
93+
"auralogger.client.client_log.threading.Thread"
94+
) as mocked_thread:
95+
client_log("info", "still prints", "tests/client", {"x": 1})
96+
97+
mocked_print.assert_called_once()
98+
mocked_thread.assert_not_called()
99+
75100
def test_close_socket_delegates(self) -> None:
76101
with patch("auralogger.client.client_log.close_client_log_socket") as mocked:
77102
AuraClient.close_socket()

0 commit comments

Comments
 (0)