INNER CODE UNIT · Python

_drain_proxy_persistence_tasks

Soju06/codex-lb · app/main.py:247

async def _drain_proxy_persistence_tasks(
    proxy_service: Any,
    timeout_seconds: float,
    *,
    task_name_prefixes: tuple[str, ...] | None = None,
    failure_message: str,
) -> bool:
    """Drain proxy persistence work within the caller's committed deadline."""
    if proxy_service is None or not hasattr(proxy_service, "drain_persistence_tasks"):
        return True
    try:
        kwargs: dict[str, Any] = {"timeout_seconds": timeout_seconds}
        if task_name_prefixes is not None:
            kwargs["task_name_prefixes"] = task_name_prefixes
        return bool(await proxy_service.drain_persistence_tasks(**kwargs))
    except Exception:
        logger.warning(failure_message, exc_info=True)
        return False

View source record →