Documents
Home>Documents>AI>Agent>Xgen

Redis Session Management with Concurrent Access Control

5 min readApr 15, 2025Feb 22, 2026

Redis Session Management — Handling Concurrent Connections

The XGen platform supports multiple users editing and running workflows simultaneously. This post documents how we introduced Redis for session management and concurrent connection handling.

Redis Connection Setup

import redis.asyncio as redis
from contextlib import asynccontextmanager

class RedisManager:
    def __init__(self):
        self.pool = None

    async def init(self, url: str = "redis://localhost:6379"):
        self.pool = redis.ConnectionPool.from_url(
            url, max_connections=50, decode_responses=True
        )

    async def close(self):
        if self.pool:
            await self.pool.disconnect()

    @property
    def client(self) -> redis.Redis:
        return redis.Redis(connection_pool=self.pool)

redis_manager = RedisManager()

Session-Based Authentication

We chose Redis sessions over JWT tokens because they allow immediate server-side invalidation and flexible storage of session data.

import secrets
from datetime import timedelta

SESSION_TTL = timedelta(hours=24)
SESSION_PREFIX = "session:"

async def create_session(user_id: str, user_data: dict) -> str:
    session_id = secrets.token_urlsafe(32)
    key = f"{SESSION_PREFIX}{session_id}"

    await redis_manager.client.hset(key, mapping={
        "user_id": user_id,
        "username": user_data["username"],
        "role": user_data["role"],
        "created_at": str(datetime.utcnow()),
    })
    await redis_manager.client.expire(key, int(SESSION_TTL.total_seconds()))

    return session_id

async def get_session(session_id: str) -> dict | None:
    key = f"{SESSION_PREFIX}{session_id}"
    data = await redis_manager.client.hgetall(key)
    if not data:
        return None
    # Refresh TTL on session access (sliding expiration)
    await redis_manager.client.expire(key, int(SESSION_TTL.total_seconds()))
    return data

FastAPI Middleware Integration

from fastapi import Request, HTTPException

async def auth_middleware(request: Request):
    session_id = request.cookies.get("session_id")
    if not session_id:
        raise HTTPException(status_code=401, detail="인증이 필요합니다")

    session = await get_session(session_id)
    if not session:
        raise HTTPException(status_code=401, detail="세션이 만료되었습니다")

    request.state.user = session
    return session

Concurrent Session Limits

We added a limit on the number of active sessions allowed per account.

MAX_SESSIONS_PER_USER = 3

async def create_session_with_limit(user_id: str, user_data: dict) -> str:
    user_sessions_key = f"user_sessions:{user_id}"

    # Check current active session count
    active_sessions = await redis_manager.client.smembers(user_sessions_key)

    # Clean up expired sessions
    for sid in active_sessions:
        exists = await redis_manager.client.exists(f"{SESSION_PREFIX}{sid}")
        if not exists:
            await redis_manager.client.srem(user_sessions_key, sid)

    active_count = await redis_manager.client.scard(user_sessions_key)
    if active_count >= MAX_SESSIONS_PER_USER:
        # Evict the oldest session
        oldest = list(active_sessions)[0]
        await redis_manager.client.delete(f"{SESSION_PREFIX}{oldest}")
        await redis_manager.client.srem(user_sessions_key, oldest)

    session_id = await create_session(user_id, user_data)
    await redis_manager.client.sadd(user_sessions_key, session_id)
    return session_id

Production Tips

Because Redis is memory-based, we used it for much more than sessions — including caching workflow execution state and storing intermediate results. Caching mid-execution results in Redis was especially useful: if an SSE connection dropped, clients could reconnect and resume from where they left off.

Tags
RedisSessionAuthenticationFastAPIConcurrent Access