""" db_client.py — single shared, hardened Supabase client. All runtime modules import `supabase` from here instead of each calling create_client() themselves. Two reasons: 1. HTTP/1.1 instead of HTTP/2. The default PostgREST httpx client uses HTTP/2, and Supabase's edge intermittently sends a graceful GOAWAY to recycle the connection. httpx then raises on the pooled connection: httpx.RemoteProtocolError: which surfaced as random 500s ("I seem to have lost my train of thought"). Forcing http2=False removes that whole error class. 2. One connection pool instead of ~10 (every module used to build its own client), with sane timeouts and short keep-alive so idle connections are dropped before the server closes them. db_execute() wraps a query with a small retry for the residual transient disconnect that HTTP/1.1 keep-alive can still hit. """ import time import httpx from supabase import create_client from supabase.lib.client_options import ClientOptions from config import SUPABASE_URL, SUPABASE_SERVICE_KEY # CRITICAL: PostgREST and Storage must NOT share one httpx.Client. # supabase-py passes a single ClientOptions.httpx_client to BOTH the postgrest and # storage sub-clients. The storage client SETS base_url=".../storage/v1/" on that # shared client the first time it is used. With a shared client, every DB query # AFTER a storage op (e.g. the db5 FAISS archive that runs on every turn) was sent # to the storage endpoint -> Fastify "Route ... not found" 404 storm on tables # that exist. Reproduced and verified locally. Fix: a separate httpx client for # storage so it cannot corrupt postgrest's base_url. # http2=False avoids the original HTTP/2 GOAWAY RemoteProtocolError. def _make_httpx() -> httpx.Client: return httpx.Client( http2=False, timeout=httpx.Timeout(30.0, connect=10.0), limits=httpx.Limits( max_keepalive_connections=10, max_connections=40, keepalive_expiry=15.0, ), ) _httpx_client = _make_httpx() # postgrest (+ auth + functions) _storage_httpx_client = _make_httpx() # storage ONLY — isolated try: import supabase as _supabase_pkg _SUPABASE_VERSION = getattr(_supabase_pkg, "__version__", "unknown") except Exception: _SUPABASE_VERSION = "unknown" try: supabase = create_client( SUPABASE_URL, SUPABASE_SERVICE_KEY, options=ClientOptions(httpx_client=_httpx_client), ) # Re-bind the storage sub-client to its OWN httpx client so storage ops can't # overwrite postgrest's base_url (see comment above). try: supabase._storage = supabase._init_storage_client( storage_url=supabase.storage_url, headers=supabase.options.headers, storage_client_timeout=supabase.options.storage_client_timeout, http_client=_storage_httpx_client, ) print(f"[db_client] supabase-py {_SUPABASE_VERSION}: HTTP/1.1 client ACTIVE, storage isolated") except Exception as _e: print(f"[db_client] WARNING: could not isolate storage client ({_e}); " "storage ops may corrupt postgrest base_url") print(f"[db_client] supabase-py {_SUPABASE_VERSION}: HTTP/1.1 client ACTIVE") except TypeError: # Older supabase-py (<2.15) has no httpx_client option. Don't crash the app — # fall back to the default client (HTTP/2, the RemoteProtocolError risk returns). # Bump the supabase pin in requirements.txt to restore the hardened client. print(f"WARNING: supabase-py {_SUPABASE_VERSION} too old for httpx_client; " "using default HTTP/2 client. Pin supabase>=2.15 to restore HTTP/1.1.") supabase = create_client(SUPABASE_URL, SUPABASE_SERVICE_KEY) # Transient transport errors worth retrying (stale keep-alive, brief network blips). _RETRYABLE = (httpx.RemoteProtocolError, httpx.ConnectError, httpx.ReadError, httpx.WriteError) try: from postgrest.exceptions import APIError except Exception: # pragma: no cover - import shape varies across versions APIError = None def _is_transient_schema_cache_error(e) -> bool: """PostgREST briefly returns 404 'Route ... not found' / PGRST205 'schema cache' while it reloads its schema (e.g. after DDL). The table exists; the request just raced the cache reload. These are safe to retry; genuine 4xx (constraint, bad column) are not. """ if APIError is None or not isinstance(e, APIError): return False code = str(getattr(e, "code", "") or "") msg = (str(getattr(e, "message", "") or "") + " " + str(getattr(e, "details", "") or "")).lower() return code in ("404", "PGRST205") or "schema cache" in msg or "not found" in msg def db_execute(query, retries: int = 2): """Execute a postgrest/supabase query builder, retrying transient transport errors and transient PostgREST schema-cache misses. Usage: db_execute(supabase.table("x").select("*").eq("id", uid)) """ last = None for attempt in range(retries + 1): try: return query.execute() except _RETRYABLE as e: last = e if attempt < retries: time.sleep(0.3 * (attempt + 1)) continue raise except Exception as e: if _is_transient_schema_cache_error(e) and attempt < retries: last = e time.sleep(0.5 * (attempt + 1)) continue raise raise last # pragma: no cover