Spaces:
Running
Running
File size: 2,720 Bytes
de28957 88cd064 ead8f13 99f7f2e de28957 ead8f13 de28957 99f7f2e ead8f13 de28957 99f7f2e ead8f13 99f7f2e 88cd064 ead8f13 99f7f2e ead8f13 de28957 99f7f2e de28957 99f7f2e de28957 99f7f2e ead8f13 de28957 ead8f13 de28957 99f7f2e de28957 88cd064 ead8f13 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | from __future__ import annotations
import asyncio
import logging
from redis.asyncio import Redis
from redis.asyncio.retry import Retry
from redis.backoff import ExponentialBackoff
from redis.exceptions import ConnectionError, TimeoutError
from langgraph.checkpoint.redis.aio import AsyncRedisSaver
from app.config import get_settings
logger = logging.getLogger(__name__)
_checkpointer_cm = None
_checkpointer: AsyncRedisSaver | None = None
_redis_client: Redis | None = None
_heartbeat_task: asyncio.Task | None = None
def _build_redis_client(redis_url: str) -> Redis:
return Redis.from_url(
redis_url,
health_check_interval=15,
socket_keepalive=True,
socket_connect_timeout=5,
socket_timeout=10,
retry_on_timeout=True,
retry_on_error=[ConnectionError, TimeoutError],
retry=Retry(ExponentialBackoff(base=0.5, cap=3), retries=3),
)
async def _redis_heartbeat(interval: int = 15):
"""Keeps the pooled Redis connection alive by sending real traffic
through it periodically — Railway's proxy kills idle connections,
and OS-level TCP keepalive alone doesn't count as activity to it."""
while True:
try:
await asyncio.sleep(interval)
if _redis_client is not None:
await _redis_client.ping()
except asyncio.CancelledError:
break
except Exception as e:
logger.warning("Redis heartbeat ping failed: %s", e)
async def get_checkpointer() -> AsyncRedisSaver:
global _checkpointer_cm, _checkpointer, _redis_client, _heartbeat_task
if _checkpointer is None:
settings = get_settings()
ttl_minutes = max(1, settings.chat_session_ttl_seconds // 60)
_redis_client = _build_redis_client(settings.redis_url)
_checkpointer_cm = AsyncRedisSaver.from_conn_string(
redis_client=_redis_client,
ttl={"default_ttl": ttl_minutes, "refresh_on_read": True},
)
_checkpointer = await _checkpointer_cm.__aenter__()
await _checkpointer.asetup()
_heartbeat_task = asyncio.create_task(_redis_heartbeat())
return _checkpointer
async def close_checkpointer() -> None:
global _checkpointer_cm, _checkpointer, _redis_client, _heartbeat_task
if _heartbeat_task is not None:
_heartbeat_task.cancel()
if _checkpointer_cm is not None:
await _checkpointer_cm.__aexit__(None, None, None)
if _redis_client is not None:
await _redis_client.aclose()
_checkpointer = None
_checkpointer_cm = None
_redis_client = None
_heartbeat_task = None |