diff --git a/nodes/src/nodes/tool_google_workspace/google_client.py b/nodes/src/nodes/tool_google_workspace/google_client.py index 74d2e357e..1c702f1d6 100644 --- a/nodes/src/nodes/tool_google_workspace/google_client.py +++ b/nodes/src/nodes/tool_google_workspace/google_client.py @@ -39,8 +39,10 @@ - scope diagnostics (:func:`token_scope_report`) shared by ``validateConfig``, ``check_connection``, and ``build_service`` so the three checks cannot drift; - request execution with exponential backoff on 429/5xx and on rate-limit - 403s (parsed from the structured error body), while permission 403s and - other errors fail fast with an agent-readable message. + 403s (parsed from the structured error body), plus the same backoff for + status-less transport faults on GET requests only, all on a per-thread + transport (httplib2 is not thread safe under parallel agent tool calls); + permission 403s and other errors fail fast with an agent-readable message. Per-service subpackages keep only their tool functions and response cleaners, and bind these functions via ``functools.partial`` (see e.g. ``sheets/client.py``). @@ -52,6 +54,7 @@ import binascii import json import os +import threading import time as _time from dataclasses import dataclass, field from typing import Any @@ -399,23 +402,68 @@ def build_service(svc: GoogleService, auth_type: str, cfg: dict, scopes: list[st return build(svc.api, svc.version, credentials=creds, cache_discovery=False) +_thread_transport = threading.local() + + +def _request_http(request: Any): + """Per-thread transport for a request built on a shared service handle. + + googleapiclient services carry one httplib2 transport and every request + built from a service inherits it. httplib2 is not thread safe: concurrent + ``execute()`` calls on the same handle can interleave on one TLS + connection, which surfaces as ``[SSL] record layer failure`` when an agent + executor runs a wave of tool calls in parallel threads. Rebuild the + authorized transport once per thread, keyed by the credential object, so + parallel calls never share a connection. Returns None (keep the request's + own transport) when there is nothing to rebuild from. + """ + shared = getattr(request, 'http', None) + creds = getattr(shared, 'credentials', None) + if creds is None: + return None + try: + import google_auth_httplib2 + import httplib2 + except ImportError: + return None + cache = getattr(_thread_transport, 'by_creds', None) + if cache is None: + cache = {} + _thread_transport.by_creds = cache + http = cache.get(id(creds)) + if http is None: + http = google_auth_httplib2.AuthorizedHttp(creds, http=httplib2.Http()) + cache[id(creds)] = http + return http + + def execute(svc: GoogleService, request: Any, *, binary: bool = False) -> Any: """Run an API request with exponential backoff on 429/5xx and rate-limit 403s. + Status-less failures (connection reset, TLS fault, timeout) are retried + with the same backoff for GET requests only. A mutation whose response was + lost may still have landed on Google's side, so it fails fast and leaves + the re-check to the caller rather than risking a duplicated side effect. + Requests run on a per-thread transport (see ``_request_http``). + ``binary=True`` returns the raw response (bytes from get_media/export_media); otherwise the JSON dict (or {} when the API returns no body). With 4 attempts the sleeps are 1s, 2s, 4s — worst case ~7s before the final raise. """ base_delay = 1.0 + http = _request_http(request) for attempt in range(4): try: - result = request.execute() + result = request.execute(http=http) if http is not None else request.execute() return result if binary else (result or {}) except Exception as exc: # googleapiclient.errors.HttpError and transport errors status = getattr(getattr(exc, 'resp', None), 'status', None) if status and (int(status) in _RETRY_STATUSES or _is_rate_limit_403(exc)) and attempt < 3: _time.sleep(base_delay * (2**attempt)) continue + if status is None and attempt < 3 and getattr(request, 'method', None) == 'GET': + _time.sleep(base_delay * (2**attempt)) + continue detail = getattr(exc, 'reason', None) or str(exc) if status and int(status) == 403: raise ValueError( diff --git a/nodes/test/tool_google_workspace/test_google_client.py b/nodes/test/tool_google_workspace/test_google_client.py index 071aa182c..42a007c2a 100644 --- a/nodes/test/tool_google_workspace/test_google_client.py +++ b/nodes/test/tool_google_workspace/test_google_client.py @@ -294,3 +294,89 @@ def read(self): creds.refresh(None) # the half-update is the bug: neither token nor expiry may have changed assert creds.token == 'stale' + + +def _scripted_request(method, outcomes): + """Request double whose execute() pops scripted outcomes in order.""" + calls = {'count': 0, 'http_seen': []} + + def execute(http=None): + calls['http_seen'].append(http) + index = min(calls['count'], len(outcomes) - 1) + calls['count'] += 1 + outcome = outcomes[index] + if isinstance(outcome, Exception): + raise outcome + return outcome + + request = types.SimpleNamespace(method=method, http=None, execute=execute) + return request, calls + + +def test_execute_retries_transport_faults_for_get(monkeypatch, service): + monkeypatch.setattr(google_client._time, 'sleep', lambda seconds: None) + fault = OSError('[SSL] record layer failure (_ssl.c:2580)') + request, calls = _scripted_request('GET', [fault, fault, {'ok': True}]) + + assert google_client.execute(service, request) == {'ok': True} + assert calls['count'] == 3 + + +def test_execute_gives_up_on_get_after_four_transport_attempts(monkeypatch, service): + monkeypatch.setattr(google_client._time, 'sleep', lambda seconds: None) + fault = OSError('connection reset by peer') + request, calls = _scripted_request('GET', [fault]) + + with pytest.raises(ValueError, match='request failed'): + google_client.execute(service, request) + assert calls['count'] == 4 + + +def test_execute_does_not_retry_transport_faults_for_mutations(monkeypatch, service): + monkeypatch.setattr(google_client._time, 'sleep', lambda seconds: None) + request, calls = _scripted_request('POST', [OSError('connection reset by peer')]) + + with pytest.raises(ValueError, match='request failed'): + google_client.execute(service, request) + assert calls['count'] == 1 + + +def test_execute_uses_a_distinct_transport_per_thread(monkeypatch, service): + import threading + + class FakeAuthorizedHttp: + def __init__(self, credentials, http=None): + self.credentials = credentials + self.http = http + + fake_google_auth = types.SimpleNamespace(AuthorizedHttp=FakeAuthorizedHttp) + fake_httplib2 = types.SimpleNamespace(Http=lambda: object()) + monkeypatch.setitem(sys.modules, 'google_auth_httplib2', fake_google_auth) + monkeypatch.setitem(sys.modules, 'httplib2', fake_httplib2) + monkeypatch.setattr(google_client, '_thread_transport', threading.local()) + + credentials = object() + per_thread = {} + + def run(name): + seen = [] + for _ in range(2): + request = types.SimpleNamespace( + method='GET', + http=types.SimpleNamespace(credentials=credentials), + execute=lambda http=None: seen.append(http) or {}, + ) + google_client.execute(service, request) + per_thread[name] = seen + + threads = [threading.Thread(target=run, args=(i,)) for i in range(2)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + first, second = per_thread[0], per_thread[1] + assert first[0] is first[1], 'a thread must reuse its own transport' + assert second[0] is second[1], 'a thread must reuse its own transport' + assert first[0] is not second[0], 'threads must not share a transport' + assert isinstance(first[0], FakeAuthorizedHttp)