feat(portaswitch): make async HTTP connection-pool size configurable (WT-1720)

Follow-up to the sync->async migration. The httpx.AsyncClient connection pool —
not the former ~40 Starlette thread-pool tokens — is now the real per-pod
ceiling on concurrent requests toward the switch, so expose its size via env:

- PORTASWITCH_MAX_CONNECTIONS (default 100)
- PORTASWITCH_MAX_KEEPALIVE_CONNECTIONS (default 20)

Defaults match httpx's own: an order of magnitude above the old thread cap while
staying polite to the switch. Raise for a large/dedicated switch or high Cloud
Run concurrency; lower to protect a small or shared one. Blank/invalid env
values fall back to the defaults. Applied to the shared AsyncClient via
httpx.Limits; http_api.py and the other adapters are unaffected.

Author: yakimenko73

app/bss/adapters/portaswitch/api/account.py

@@ -28,6 +28,9 @@ def __init__(self, portaswitch_settings: PortaSwitchSettings):
         self._verify_https = portaswitch_settings.VERIFY_HTTPS
         if portaswitch_settings.API_TIMEOUT is not None:
             self.DEFAULT_REQUEST_TIMEOUT = portaswitch_settings.API_TIMEOUT
+        # httpx connection-pool limits for the shared async client (WT-1720).
+        self._max_connections = portaswitch_settings.MAX_CONNECTIONS
+        self._max_keepalive_connections = portaswitch_settings.MAX_KEEPALIVE_CONNECTIONS
 
     async def __send_request(
             self, module: str, method: str, params: dict, stream: bool | None = None, access_token: str | None = None

app/bss/adapters/portaswitch/api/admin.py

@@ -31,6 +31,9 @@ def __init__(self, portaswitch_settings: PortaSwitchSettings) -> None:
         self._verify_https = portaswitch_settings.VERIFY_HTTPS
         if portaswitch_settings.API_TIMEOUT is not None:
             self.DEFAULT_REQUEST_TIMEOUT = portaswitch_settings.API_TIMEOUT
+        # httpx connection-pool limits for the shared async client (WT-1720).
+        self._max_connections = portaswitch_settings.MAX_CONNECTIONS
+        self._max_keepalive_connections = portaswitch_settings.MAX_KEEPALIVE_CONNECTIONS
         self._api_user = PortaSwitchAdminUser(
             user_id=portaswitch_settings.ADMIN_API_LOGIN, token=portaswitch_settings.ADMIN_API_TOKEN
         )

app/bss/adapters/portaswitch/config.py

@@ -35,6 +35,16 @@ class PortaSwitchSettings(BaseSettings):
     # slow/unresponsive switch can't pin worker threads indefinitely (WT-1717).
     # Configurable via PORTASWITCH_API_TIMEOUT; set empty to keep the base default.
     API_TIMEOUT: Optional[float] = 25
+    # httpx connection-pool limits for the async client (WT-1720). This pool — not
+    # the old ~40 Starlette thread-pool tokens — is now the real per-pod ceiling on
+    # concurrent requests toward the switch. Defaults match httpx's own (100/20):
+    # an order of magnitude above the former thread cap, while staying polite to the
+    # switch. Raise MAX_CONNECTIONS for a large switch / high Cloud Run concurrency;
+    # lower it to protect a small or shared one. Keep MAX_KEEPALIVE_CONNECTIONS
+    # <= MAX_CONNECTIONS. Configurable via PORTASWITCH_MAX_CONNECTIONS /
+    # PORTASWITCH_MAX_KEEPALIVE_CONNECTIONS.
+    MAX_CONNECTIONS: int = 100
+    MAX_KEEPALIVE_CONNECTIONS: int = 20
     SIGNIN_CREDENTIALS: PortaSwitchSignInCredentialsType = PortaSwitchSignInCredentialsType.SELF_CARE
     CONTACTS_SELECTING: PortaSwitchContactsSelectingMode = PortaSwitchContactsSelectingMode.ACCOUNTS
     CONTACTS_SELECTING_EXTENSION_TYPES: Union[List[PortaSwitchExtensionType], str] = list(PortaSwitchExtensionType)
@@ -60,6 +70,28 @@ def decode_api_timeout(cls, v: Union[str, float, int, None]) -> Optional[float]:
             return None
         return v if v > 0 else None
 
+    @staticmethod
+    def _positive_int_or(v: Union[str, int, None], default: int) -> int:
+        # Treat blank/invalid/non-positive as "unset" so a stray empty env var
+        # (e.g. PORTASWITCH_MAX_CONNECTIONS="") falls back to the safe default.
+        if v is None or (isinstance(v, str) and not v.strip()):
+            return default
+        try:
+            iv = int(v)
+        except (TypeError, ValueError):
+            return default
+        return iv if iv > 0 else default
+
+    @field_validator("MAX_CONNECTIONS", mode='before')
+    @classmethod
+    def decode_max_connections(cls, v: Union[str, int, None]) -> int:
+        return cls._positive_int_or(v, 100)
+
+    @field_validator("MAX_KEEPALIVE_CONNECTIONS", mode='before')
+    @classmethod
+    def decode_max_keepalive_connections(cls, v: Union[str, int, None]) -> int:
+        return cls._positive_int_or(v, 20)
+
     @field_validator("CONTACTS_SELECTING_EXTENSION_TYPES", mode='before')
     @classmethod
     def decode_contacts_selecting_extension_types(cls, v: Union[List, str]) -> List[PortaSwitchExtensionType]:

app/bss/async_http_api.py

@@ -77,16 +77,25 @@ def build_httpx_timeout(value) -> httpx.Timeout:
 _shared_clients_lock = asyncio.Lock()
 
 
-async def get_shared_async_client(verify: bool) -> httpx.AsyncClient:
+async def get_shared_async_client(verify: bool, limits: Optional[httpx.Limits] = None) -> httpx.AsyncClient:
     """Return the process-wide :class:`httpx.AsyncClient` for the given TLS
-    verify setting, creating it lazily inside the running event loop."""
+    verify setting, creating it lazily inside the running event loop.
+
+    ``limits`` (connection-pool sizing) is applied only when the client is first
+    created for a given verify key; a later caller with different limits reuses
+    the existing client. In practice all connectors in one deployment pass the
+    same limits, so this is not a concern.
+    """
     client = _shared_clients.get(verify)
     if client is not None and not client.is_closed:
         return client
     async with _shared_clients_lock:
         client = _shared_clients.get(verify)
         if client is None or client.is_closed:
-            client = httpx.AsyncClient(verify=verify)
+            kwargs = {"verify": verify}
+            if limits is not None:
+                kwargs["limits"] = limits
+            client = httpx.AsyncClient(**kwargs)
             _shared_clients[verify] = client
         return client
 
@@ -119,8 +128,17 @@ def _verify(self) -> bool:
     def _request_timeout(self) -> httpx.Timeout:
         return build_httpx_timeout(self.DEFAULT_REQUEST_TIMEOUT)
 
+    def _limits(self) -> Optional[httpx.Limits]:
+        # Subclasses set _max_connections / _max_keepalive_connections after
+        # super().__init__(); returns None to keep httpx's own defaults.
+        mc = getattr(self, "_max_connections", None)
+        mk = getattr(self, "_max_keepalive_connections", None)
+        if mc is None and mk is None:
+            return None
+        return httpx.Limits(max_connections=mc, max_keepalive_connections=mk)
+
     async def _client(self) -> httpx.AsyncClient:
-        return await get_shared_async_client(self._verify())
+        return await get_shared_async_client(self._verify(), self._limits())
 
     def add_auth_info(self, url: str, request_params: dict,
                       auth_session: AuthSessionData) -> dict: