跪拜 Guibai
← Back to the summary

How Two Async Bugs in Redis Session Storage Wiped 40-Minute AI Conversations

At 2 a.m., a user ticket jolted me awake: "I chatted with the AI about requirements for 40 minutes tonight, refreshed the page midway, and it forgot everything we'd discussed, making me introduce myself all over again." My first reaction was that the data must be in Redis; it worked fine in the dev environment. But reproducing this kind of "intermittent amnesia" manually is pure luck — until I wrote a Playwright end-to-end test, ran it 300 times, and the bug that had been hiding for three days finally surfaced.

Problem Breakdown

Implementing memory persistence for large language models is straightforward: store the conversation history, and when the user returns, include the historical messages to avoid context breaks. We used the classic approach of binding a session_id to a conversation list. After receiving a message, the backend pushes it into a Redis list with a 7-day expiration. On the surface, it looked flawless.

But the user feedback had two anomalies:

  1. It didn't happen every time: Most refreshes were fine, but occasionally the entire conversation would vanish, as if it had been wiped with a single click.
  2. It concentrated on long conversations: Tests with just a few messages and a refresh never failed, but after a dozen or more rounds, refreshing had a high probability of amnesia.

Conventional unit tests only covered the "store it, read it back" path, completely missing the user's real operational sequence: send a message → wait for a reply → page renders → user refreshes → reload. And manual testing simply couldn't hit the critical window where "the async write hasn't finished" or "Redis just expired."

What I lacked wasn't another caching strategy, but an automated test that could precisely replay the user's chat sequence and freely control time and page lifecycle.

Solution Design

There are plenty of tools on the market that can drive a real browser: Selenium, Cypress, Playwright. Why choose Playwright?

The overall idea: use pytest-playwright to write a parameterized test case simulating "multiple rounds of conversation → forced refresh → check if memory persists," and perform full-chain assertions on the backend service's async writes and Redis expiration policy. If history loss is detected during the test, immediately dump page screenshots and network logs for investigation.

Core Implementation

To reproduce the problem, I first built a minimal chat backend exposing two endpoints: POST /chat to send a message and return an AI reply, and GET /history?session_id=xxx to retrieve the complete conversation history for that session. Memory is stored in Redis, and session_id is maintained via a cookie. Here's the buggy backend code (with bugs included):

# app.py - Buggy backend, solely for reproducing the bug
import asyncio
import uuid
from flask import Flask, request, jsonify, make_response
import redis.asyncio as redis

app = Flask(__name__)
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

async def save_message(session_id, role, content):
    # Simulate async write for LLM-generated reply
    await r.rpush(f"chat:{session_id}", f"{role}:{content}")
    # Set expiration once after writing, but there's no renewal logic here
    await r.expire(f"chat:{session_id}", 7*24*3600)

@app.route('/chat', methods=['POST'])
async def chat():
    data = request.json
    session_id = request.cookies.get('session_id', str(uuid.uuid4()))
    user_msg = data['message']
    # Store user message — note: no await here!
    asyncio.create_task(save_message(session_id, 'user', user_msg))
    # Simulate AI reply (simplified to a fixed response)
    bot_reply = f"You said: {user_msg}, I remember we chatted before." if await r.llen(f"chat:{session_id}") > 1 else "Please start your question."
    # Similarly, don't wait for the bot message write
    asyncio.create_task(save_message(session_id, 'bot', bot_reply))
    resp = make_response(jsonify({"reply": bot_reply, "session_id": session_id}))
    resp.set_cookie('session_id', session_id)
    return resp

@app.route('/history', methods=['GET'])
async def history():
    session_id = request.cookies.get('session_id', '')
    msgs = await r.lrange(f"chat:{session_id}", 0, -1)
    return jsonify(msgs)

if __name__ == '__main__':
    app.run(port=5000)

What this code solves: It demonstrates a persistence implementation that "looks like it runs" but contains two fatal bugs. asyncio.create_task doesn't wait for the task to complete, so the main thread returns a response immediately; expire is only set once on the first write, and subsequent appended messages don't renew it. If there's even a slight network or CPU delay, Redis may not have finished writing when the user refreshes, or the key may have just expired.

Next is the Playwright test script, used to stably reproduce this "refresh loses memory" scenario:

# test_memory.py
import re
from playwright.sync_api import Page, expect
import pytest

@pytest.mark.parametrize("rounds", [5, 15])   # Short chat and long chat scenarios
def test_chat_memory_survives_refresh(page: Page, rounds: int):
    page.goto("http://localhost:5000")          # Open chat
    # Multiple rounds of conversation
    for i in range(rounds):
        msg = f"Round {i} message"
        page.fill("input#message", msg)
        page.click("button#send")
        # Wait for the AI reply to fully appear
        page.wait_for_selector(f"text=You said: {msg}", timeout=5000)
    # Key operation: forced refresh, simulating user reloading the page
    page.reload()
    # Assertion: history should contain at least the first round's user message
    first_msg = "Round 0 message"
    expect(page.locator("#history")).to_contain_text(first_msg, timeout=3000)

What this code solves: It uses Playwright to simulate real user interaction: typing a message, sending, waiting for a reply, and finally refreshing. The test parameterizes rounds — 5 rounds for short chats, 15 for long chats — precisely where the bug occurs most frequently. page.wait_for_selector ensures the backend has finished processing before moving to the next step, far more reliable than blind sleep.

Running it reveals: rounds=5 fails occasionally, rounds=15 fails almost every time. Checking Redis, the key either doesn't exist or its length is far less than expected, confirming the two issues: "write not completed" and "expiration not renewed."

The fixed backend is extremely simple: replace create_task with await, and refresh the expiration time on every write or read:

@app.route('/chat', methods=['POST'])
async def chat():
    # ... get session_id, user_msg
    await save_message(session_id, 'user', user_msg)       # Wait for it to finish writing
    bot_reply = "..."
    await save_message(session_id, 'bot', bot_reply)
    resp = make_response(...)
    resp.set_cookie('session_id', session_id)
    return resp

@app.route('/history', methods=['GET'])
async def history():
    session_id = request.cookies.get('session_id', '')
    # Renew before reading to ensure active sessions don't expire
    await r.expire(f"chat:{session_id}", 7*24*3600)
    msgs = await r.lrange(f"chat:{session_id}", 0, -1)
    return jsonify(msgs)

What this code solves: Using await ensures messages are persisted to disk before the response is returned; refreshing the TTL before each history read ensures that as long as the user is still chatting, the key won't be evicted. The subsequent Playwright tests ran 300 times, all passing.

Pitfalls Encountered

Pitfall 1: Cookie loss after Playwright's page.reload() causes session breakage

Pitfall 2: wait_for_selector text matching broken by frontend line breaks

Results Verification

Before the fix, running 300 long-chat refresh tests with Playwright resulted in only 47 passes, a failure rate of 84%. After the fix, the same 300 tests were all green, a 100% pass rate. The key change was going from "occasionally lost" to ZERO, and no more amnesia tickets ever came in.

Ready-to-Use Code/Tool

If you also use Redis to store sessions, just copy this renewal logic into your history read function:

async def get_history(session_id):
    await redis.expire(f"chat:{session_id}", 7*24*3600)
    return await redis.lrange(f"chat:{session_id}", 0, -1)

Friends working on LLM applications can take this directly — it's one line of code that will save you at least one late-night phone call.

#Playwright #LLM #AutomatedTesting #BackendPitfalls #Redis


About the Author A backend/architecture practitioner who considers "if it runs, it's fine" a disgrace, focused on using testing and tools to kill bugs at the code level. GitHub: https://github.com/baofugege Sponsor: https://github.com/sponsors/baofugege — If this article saved you troubleshooting time, buy me a coffee. Services: Python backend performance optimization / tool customization / technical consulting, contact Telegram @baofugege