Forem: BAOFUFAN

Building RAG with LangChain & Chroma: Two Hidden Pitfalls That Cost Me 6 Hours

BAOFUFAN — Wed, 06 May 2026 01:08:16 +0000

At 10 PM, my product manager dropped 200 PDFs in my lap: “We need to demo an internal knowledge base Q&A for the boss tomorrow morning—super urgent.” I thought, “RAG? I know this; LangChain plus a vector database, done in minutes.” I started coding at 4 PM and barely finished by 10 PM—not because the pipeline didn’t run, but because two subtle traps dragged the accuracy below 40% and had me debugging for six straight hours. In this article, I’ll walk you through the full RAG system build and pull the two pitfalls out by their roots.

Why you can’t just dump documents into GPT

The simplest idea for a system that answers questions like “What is the company holiday policy?” or “What were the conclusions of project X’s retrospective?” is to concatenate all the documents into one giant prompt and send it to GPT. Reality hits fast: 200 PDFs add up to over 800,000 characters. Even GPT-4’s 128K context window chokes, and the per‑call cost will make your finance team come after you. Fine‑tuning is even less realistic—the documents change daily, and you’re not going to burn thousands of dollars every time they do.

That leaves retrieval‑augmented generation (RAG) as the only viable path: split documents into small chunks, embed each chunk with an embedding model and store the vectors in a vector database. At query time, retrieve the most relevant chunks, stuff them into the prompt as context, and let the LLM generate an answer. The pattern looks simple, but every step—“how to split,” “how to store,” “how to search”—has its own sharp edges. The two that wrecked me were buried deep in the interaction between LangChain and Chroma.

Design choices: why Chroma over Pinecone or FAISS

Before picking a vector store, I asked myself three questions: does it cost money, does it support metadata filtering, and can I start/stop it locally with a single command?

Pinecone costs money and requires data to go to the cloud—immediately ruled out for internal documents. Weaviate is powerful, but deploying it means at least 30 minutes of Docker tinkering, a non‑starter when the demo is “tomorrow morning.” FAISS is blazing fast, but it doesn’t support metadata filtering (like filtering by document type or date range)—a feature we’d need as soon as the business side piles on more requirements. I landed on Chroma: it runs locally, installs with a single pip install chromadb, and has persistence, metadata filtering, and similarity search built right in. It also integrates with LangChain more smoothly than any other option.

The overall architecture is straightforward: load documents → split text → generate embeddings → write to Chroma → when a user asks, retrieve top‑k chunks → stuff into a prompt → LLM generates an answer. LangChain chains these steps together; you just need to manage the parameters and edge cases for each stage.

Core implementation: two scripts to run the full RAG pipeline

Script 1’s job: turn scattered PDFs into searchable vector chunks and persist them in Chroma so you don’t have to re‑index everything on the next run.

import os
from langchain_community.document_loaders import PyPDFDirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 1. Load PDF directory
loader = PyPDFDirectoryLoader("./docs")          # Auto-scan all PDFs
documents = loader.load()
print(f"Loaded documents: {len(documents)} pages")

# 2. Split: chunk_size and overlap are the source of two major pitfalls, detailed later
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,         # Max characters per chunk
    chunk_overlap=200,       # Overlap to avoid cutting key info across boundaries
    separators=["\n\n", "\n", "。", "，", " ", ""]  # Priority: paragraphs first
)
chunks = text_splitter.split_documents(documents)
print(f"Total chunks: {len(chunks)}")

# 3. Generate embeddings and store in Chroma (auto-persist to local dir)
embeddings = OpenAIEmbeddings()                  # Default: text-embedding-ada-002
vectordb = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"              # Reusable after restart, saves re-embedding cost
)
vectordb.persist()
print("Vector store built and persisted to ./chroma_db")

Script 2’s job: using the stored vector store, build the full “ask → retrieve → generate” chain and force the LLM to answer strictly from the provided documents—no hallucinations.

from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# 1. Custom prompt: force LLM to base answers only on the given context
prompt_template = """You are a rigorous internal knowledge base assistant. Answer the question strictly based on the context below.
If the answer cannot be found in the context, simply say "No relevant information found." Do not make anything up.

Context:
{context}

Question: {question}
Answer:"""
PROMPT = PromptTemplate(
    template=prompt_template, 
    input_variables=["context", "question"]
)

# 2. Load the persisted vector store, connecting to the same embedding model
vectordb = Chroma(
    persist_directory="./chroma_db",
    embedding_function=OpenAIEmbeddings()
)

# 3. Create the QA chain: retrieves top-4 chunks by default, using our custom prompt
llm = ChatOpenAI(model_name="gpt-4o", temperature=0)  # temperature=0 for deterministic results
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",                               # Stuffs retrieved chunks directly into the prompt
    retriever=vectordb.as_r

Note: The final line of the code block is shown exactly as it appears in the original article.

How I Cut LLM Memory Bug Diagnosis from 2 Hours to 5 Minutes with Playwright & Allure

BAOFUFAN — Tue, 05 May 2026 12:17:50 +0000

At 2 a.m., the customer group chat erupted: "Your AI assistant completely forgot the client background I provided last week and even fabricated new details!" I rolled out of bed, opened the logs, and stared at tens of thousands of conversation lines—like finding a needle in a haystack. That night, I spent nearly two hours tracking down the cause: when conversations exceeded 24 turns, our history truncation logic silently dropped the system prompts in the middle, causing memory to break completely. The next day, I decided this torture had to end with automation. If you're building LLM applications and are tormented by "occasional amnesia" or "intermittent hallucinations," this article will hand you a ready-to-use end-to-end memory testing solution.

Problem Breakdown: LLM "Amnesia" Is Scarier Than Hallucination

Once an LLM application is live, users assume it will remember conversation history. The reality is: any link in the chain involving context window management, RAG retrieval, or multi-agent state passing can cause "amnesia"—needed information fails to reach the model, or gets truncated and overwritten along the way. The risk with such bugs is that they don't fail every time; they typically surface only under specific conversation lengths or message sequences, making manual reproduction extremely hard.

Conventional testing methods fall into two buckets: (1) Unit tests on the model API, verifying single-turn input/output, never touching memory logic. (2) Manually clicking through dozens of conversation turns on the UI, time-consuming, exhausting, and impossible to precisely replicate the exact path. One forgotten step by the tester and the bug slips away. What we truly lack is an automated approach that can converse with the system like a real user and clearly record what it remembers and at which turn it starts forgetting.

Solution Design: Why Playwright + Allure?

Facing this requirement, I evaluated several paths:

Direct API calls: Fast, but bypass front-end logic like message assembly, timestamps, and user identity injection, missing the full real-world chain.
Selenium: Mature ecosystem, but async waiting and Shadow DOM handling in modern front-end frameworks are painful, and the community has clearly shifted toward Playwright.
Cypress: Limited to JS/TS ecosystem. Our backend and model services are Python-based, making stack unification too costly.

The reasons for finally choosing Playwright are straightforward: native sync/async support, auto-wait mechanism drastically reducing flaky tests, and perfect control over Chromium to capture screenshots of every critical step. Allure was chosen because its reports come with built-in step trees, attachment display, and failure highlighting, enabling non-technical stakeholders (like product managers) to instantly grasp at which turn the memory broke.

The architecture is simple: pytest + playwright + allure-pytest. Test cases use Playwright to operate the live front-end, execute multiple conversation turns, and assert each turn that key information is still remembered by the model. Allure packages the conversation content, model responses, assertion results, and page screenshots for each round into a single HTML report, making the reproduction path crystal clear.

Core Implementation: From Opening the Browser to Automated Memory Assertions

First Code Block: Solving Browser Reuse and Login State

Often tests slow down drastically due to login flows, requiring scanning QR codes or entering verification codes from scratch each time. We directly save the logged-in storage_state and reuse it.


python
# conftest.py - 复用登录态，避免每次测试都登录
import pytest
from playwright.sync_api import sync_playwright
from pathlib import Path

STATE_FILE = Path(__file__).parent / "auth_state.json"

@pytest.fixture(scope="session")
def browser():
    with sync_playwright() as p:
        # 如果想看到执行过程可设为 False
        browser = p.chromium.launch(headless=True, slow_m

A 2 AM Serialization Bug in LangChain Memory — And How pytest Stopped It Forever

BAOFUFAN — Tue, 05 May 2026 01:07:40 +0000

At 2:17 AM, my monitoring alert yanked me out of sleep: the customer service bot had suddenly lost its memory. Users were asking “Where is my order?” three times in a row, and it kept asking for their phone number as if they were complete strangers. I opened the logs and saw that ConversationBufferMemory was loading empty message lists. The key was still there in Redis, but somehow deserialization had silently swallowed the data. I rolled back the code from my bed and spent three hours tracing the root cause — a LangChain upgrade had introduced a pickle deserialization incompatibility that dropped entire conversation histories. Manual testing had never covered version upgrade scenarios. The next morning I made a decision: automate the integrity and performance checks for memory storage with pytest, and never let a serialization regression slip through again. Since then, regressions that used to take 30 minutes to verify now finish in 8 seconds.

The Breakdown

Our architecture was straightforward: ConversationBufferMemory + RedisChatMessageHistory persisting user sessions to Redis. Under the hood, LangChain used pickle to dump the message list into bytes and stored them under a {session_id} key — reloading it later with a simple load.

The problem hit during a version upgrade: we moved from langchain==0.0.352 to 0.1.0, and the fully qualified class names of HumanMessage and AIMessage changed. When the old pickle payload was loaded, it threw an AttributeError. Even worse, the messages property of RedisChatMessageHistory was catching that exception and silently returning an empty list — making it look like an innocent empty conversation with no errors anywhere. These kinds of bugs have two nasty traits:

Delayed impact: the blow-up doesn’t happen at upgrade time, but only when a user actually reads or writes memory again — monitoring can barely spot it in the first place.
Untestable by hand: before an upgrade, QA only validates “can we store and read” with the current version; nobody intentionally seeds old serialized data to check backward compatibility.

Conventional click-and-hope manual testing stands no chance against regressions like this. What we needed was an automated, repeatable integration suite covering:

Write → restart → read integrity
Multi-version serialized data compatibility
Performance under large message volumes
Correctness under concurrent reads and writes

The Plan

I chose pytest as the test framework. It wasn’t that unittest couldn’t do the job — but the things I needed were just too painful there:

Isolation: I wanted a painless Redis substitute. fakeredis perfectly simulates Redis commands, and combined with pytest fixtures it gives zero external dependencies during testing.
Parametrization: @pytest.mark.parametrize can cover 10, 100, or 1000 messages in a single line — no manual loops required.
Performance benchmarks: the pytest-benchmark plugin directly measures average and max latency, much more reliable than me sprinkling time.perf_counter() around.
Concurrency simulation: writing fixtures with threading or asyncio is far more intuitive than unittest’s subTest.

The overall approach: define a FakeRedis fixture in conftest.py and monkeypatch redis.Redis.from_url so that every LangChain Redis call hits the in-memory implementation transparently. Then split the tests into modules:

test_integrity.py: verify store / retrieve consistency and cross-instance loading
test_compatibility.py: simulate old serialized payloads and test migration / downgrade logic
test_performance.py: use pytest-benchmark to measure read/write ceilings
test_concurrency.py: multiple threads appending to the same memory, checking for data loss

Why not use a real Redis? A real instance is essential for CI smoke tests, but hitting one on every push during development is slow and messy. FakeRedis lets the whole suite run in just a few hundred milliseconds. That frictionless experience is what makes the team actually want to write tests — zero friction.

Core Implementation

1. conftest: Hijacking LangChain’s Redis connection with FakeRedis

The central idea is: every test shares one in-memory Redis, completely transparent to LangChain. We monkeypatch both redis.Redis.from_url and the direct constructor, so no matter how RedisChatMessageHistory creates a client, it always lands on the same FakeRedis instance.

# conftest.py
import pytest
import redis
from fakeredis import FakeRedis

@pytest.fixture(scope="function")
def fake_redis():
    """每个测试函数独立的 FakeRedis 实例"""
    return FakeRedis()

@pytest.fixture(autouse=True)
def patch_redis(monkeypatch, fake_redis):
    """将所有对 Redis 的调用劫持到 FakeRedis，实现零外部依赖"""
    # 劫持 from_url 方法，LangChain 内部用这个创建连接
    def _fake_from_url(url: str, **kwargs):
        return fake_redis

    monkeypatch.setattr(redis.Redis, "from_url", _fake_from_url)
    # 如果有地方直接 redis.Redis(...)，也一并拦截
    monkeypatch.setattr(redis, "Redis", lambda *a, **kw: fake_redis)
    return fake_redis

With this in place, any test that uses the patch_redis fixture automatically forces LangChain to read and write my isolated FakeRedis — the database is always pristine.

2. Testing integrity: write → reload must match bit for bit

Below is test_integrity.py. It verifies the most fundamental contract: whatever I store for a session must be returned exactly the same when loaded later. I parametrized it to cover single messages, medium-sized conversations, and massive message batches — all edge cases handled.

# test_integrity.py
import pytest
from langchain.memory import ConversationBufferMemory
from langchain.memory.chat_message_histories import RedisChatMessageHistory
from langchain.schema import HumanMessage, AIMessage

# ... test functions covering write/read integrity

The full suite (including the compatibility, performance, and concurrency modules) now lives in our CI pipeline. FakeRedis lets us run everything instantly, and the moment anyone bumps a LangChain version we catch serialization regressions before they ever reach production. Since that 2 AM wake-up call, we haven’t lost a single conversation to a silent pickle bug again.

"How a Refresh Wiped Out 237 Drafts — and How We Used Playwright to Stop It Forever"

BAOFUFAN — Mon, 04 May 2026 12:08:35 +0000

At 2 AM, I was jolted awake by a call from operations. Our user community was on fire: someone had spent half an hour filling out a complex form, accidentally hit refresh, and all their drafts vanished. A backend check showed 237 drafts reduced to just three. The backend wasn't to blame — the database never received a single request. The culprit was our frontend memory storage, and we had zero automated tests covering it. Later, we added automated tests for localStorage and IndexedDB persistence using Playwright, and the same kind of incident never happened again. In this article, I'll walk you through the complete "memory storage health check" blueprint — code included.

Breaking Down the Problem: Why Memory Storage Fails Silently

Frontend memory storage broadly refers to using localStorage, sessionStorage, IndexedDB, or state management persistence (like Pinia persist plugins) to cache user input so data isn't lost when the page is refreshed or closed. It’s what lets you hit F5 by accident and still see what you were typing — a baseline experience for any modern web app.

But its fragility is often underestimated. In our incident, the root cause was simple: a code refactor changed the serialization key for drafts. The old key was no longer read, so after a refresh the app assumed “no draft exists” and wrote an empty state, wiping everything. Conventional manual testing never covered this path because testers always fill forms from scratch — they don't deliberately refresh a half-filled form and then check for restoration.

Why common approaches fell short:

Unit tests: mocking localStorage can’t replicate real browser storage behavior, storage quotas, or serialization quirks.
E2E tests (Cypress/Playwright): they typically follow the “happy path” — fill from a blank state and submit — without intentionally triggering refresh or crash-recovery scenarios.
Manual verification: you can’t guarantee someone clears storage, fills half a form, refreshes, and validates restoration on every regression cycle. It’s too costly and easy to miss.

We needed an automated, repeatable test suite that could assert storage contents — something purpose-built to guard memory storage reliability.

Designing the Solution: Why Playwright for Memory Storage Testing

We evaluated three approaches:

Browser extensions + script injection: too hacky, impossible to integrate into CI, and can’t accurately simulate real user journeys.
Cypress APIs like cy.clearLocalStorage(): they can manipulate storage, but IndexedDB support is weaker, and the execution model doesn’t produce a truly “native” refresh scenario.
Playwright: native support for multiple browsers, isolated contexts, page.evaluate() to read localStorage/IndexedDB directly, and page.reload() that triggers a true page refresh. Plus, the test scripts are plain Node.js, seamlessly pluggable into existing CI pipelines.

Our final decision: write dedicated “memory storage regression cases” with Playwright, simulating the core path “fill → refresh → verify restoration” and running them automatically on every test pass.

The architecture is straightforward:

Each test case creates an isolated browser context (browser.newContext()) to guarantee a clean storage environment.
Each case follows three steps: write (simulate user input triggering auto-save) → refresh → read (assert that stored drafts match the form values after reload).
We designed reusable assertion helpers for both localStorage and IndexedDB.

Core Implementation: A Full-Body Checkup for Memory Storage

Let’s build a runnable memory storage test suite with Playwright. The core problem it solves: verifying that after a user fills out a form and it’s auto-saved to localStorage, the draft survives a page refresh intact.

1. Simulated Page Logic

To make the test runnable, here’s a minimal HTML page with auto-save to localStorage and restoration on load. Save it as test-app/index.html:

<!DOCTYPE html>
<html>
<body>
  <form id="draftForm">
    <input id="title" placeholder="标题" />
    <textarea id="content" placeholder="内容"></textarea>
  </form>
  <script>
    const form = document.getElementById('draftForm');
    const title = document.getElementById('title');
    const content = document.getElementById('content');
    const STORAGE_KEY = 'draft_v1';

    // 页面加载时恢复草稿
    function restore() {
      try {
        const saved = localStorage.getItem(STORAGE_KEY);
        if (saved) {
          const draft = JSON.parse(saved);
          title.value = draft.title || '';
          content.value = draft.content || '';
        }
      } catch (e) {}
    }

    // 输入变化自动保存
    function autoSave() {
      const draft = { title: title.value, content: content.value };
      localStorage.setItem(STORAGE_KEY, JSON.stringify(draft));
    }

    title.addEventListener('input', autoSave);
    content.addEventListener('input', autoSave);
    restore(); // 立刻恢复
  </script>
</body>
</html>

2. Playwright Test: Verifying Drafts Survive a Refresh

Install Playwright: npm i -D playwright @playwright/test

Create a playwright.config.ts that points to the test directory and the base URL.

The test below directly verifies refresh recovery — if this case fails, your live memory storage is broken.

// tests/memory-storage.spec.ts
import { test, expect } from '@playwright/test';

const DRAFT_TEXT = '这是一段重要的草稿内容，不能丢';

test.describe('记忆存储 - 草稿恢复', () => {
  test.beforeEach(async ({ page

I Spent 6 Hours Fixing LangChain's ConversationBufferMemory — Here's the Automated Test You Need

BAOFUFAN — Mon, 04 May 2026 01:07:14 +0000

At 4:59 PM on a Friday, I was about to close my laptop and sneak out when the QA colleague's icon flashed on DingTalk: "Come check this out. The support bot remembers I'm Zhang San, but when I ask for my order number, it insists it belongs to Li Si." I pulled up the logs and saw LangChain's ConversationBufferMemory behaving like it had severe amnesia — Session A was mixing up chat history from Session B. In that moment, I knew that unless I built an automated test suite to lock down the accuracy and consistency of memory storage, the next blow-up would definitely happen at 2 AM.

Breaking Down the Problem

In LLM-powered chat products, the memory module is responsible for remembering context across multiple turns — so that when the user said earlier "I live in Beijing," the weather query later can automatically include "Beijing." Sounds easy, but things get messy once you land in LangChain: ConversationBufferMemory stores all conversations in plain text. It works fine as long as the memory fits in RAM, but switch to Redis or a database for persistence, and a whole bunch of issues bubble up — serialization/deserialization, concurrent reads/writes, and trimming old messages.

In our production scenario, a customer service bot handled hundreds of concurrent users. Each user session was independent but they all shared a common Redis instance. When we first launched, QA manually tested a dozen typical conversation paths and found absolutely no cross-session memory leaks, because manual testing simply can't cover race conditions under high concurrency, nor reproduce edge cases where trim_messages mixes up adjacent sessions when a Redis connection blinks out. Once real traffic hit, bugs popped up like whack-a-mole — you fixed one, another sprang out. We desperately needed a set of regression tests that could directly verify memory read/write accuracy and cross-session isolation.

Designing the Solution

The goal was clear: run through the core logic of the memory module right in our local CI, without a real LLM or a real Redis instance, and catch issues before any code landed.

Framework choice was a no-brainer — Pytest. Its fixture capabilities are perfect for assembling different memory instances. LangChain's memory abstraction is fairly clean: BaseChatMemory provides uniform save_context and load_memory_variables interfaces, so we could write the same set of tests against different memory backends. A real Redis is too heavy, so we chose fakeredis to simulate a Redis instance in memory — quick to spin up and zero side effects. All LLM calls were banished with unittest.mock, because we were testing memory, not the LLM.

Why not use the built-in langchain.tests? They only cover the shallowest of interfaces, none of the hard-won scenarios like message type conversion or multi-session isolation. We also didn't want to run Redis in a Docker container — our CI resources are already stretched thin; adding one more container would jam the build queue by an extra 3 minutes.

The overall architecture: define a fake_redis_memory fixture inside Pytest's conftest.py, use it to construct different Memory subclasses (ConversationBufferMemory, ConversationSummaryMemory), simulate multi-turn conversations with helper functions, and then assert that the history returned by load_memory_variables is both complete and free of cross-session contamination.

Core Implementation

1. Building a Zero-Dependency Test Harness

This snippet packages fakeredis, mock LLM, and Memory instantiation into a fixture. All subsequent test cases run on top of it. The non-negotiable requirement: zero network requests, and any single test completes in under 0.3 seconds.

# conftest.py
import pytest
from unittest.mock import MagicMock
from langchain.memory import ConversationBufferMemory
from langchain_community.chat_message_histories import RedisChatMessageHistory
from fakeredis import FakeRedis

@pytest.fixture
def fake_redis_memory():
    # 用 fakeredis 构建一个假 Redis 客户端
    fake_redis_client = FakeRedis()

    def _create_memory(session_id: str):
        # 注入伪造的 Redis，保证每次测试的 session 隔离
        history = RedisChatMessageHistory(
            session_id=session_id,
            redis_client=fake_redis_client
        )
        # ConversationBufferMemory 默认 return_messages=True 时，会返回 Message 对象
        memory = ConversationBufferMemory(
            chat_memory=history,
            return_messages=True  # 关键：确保拿到结构化消息，方便断言
        )
        return memory

    return _create_memory

2. Testing Accuracy: Every Message Written Must Come Back

This test simulates two rounds of conversation and verifies that the history returned by load_memory_variables has the exact length and content we expect. It puts an end to the mysterious "I stored two lines but only got one back" bug.


python
# test_memory_accuracy.py
from langchain.schema import HumanMessage, AIMessage

def test_buffer_memory_keeps_all_messages(fake_redis_memory):
    memory = fake_redis_memory("session_1202")

    # 模拟第一轮对话
    memory.save_context(
        {"input": "我叫张三"},
        {"output": "你好张三"}
    )
    # 模拟第二轮对话
    memory.save_context(
        {"input": "我的订单号是多少"},
        {"output": "你的订单号是 #1123"}
    )

    variables = memory.load_memory_variables({})
    history = variables.get("history", [])

    # 断言：总共应该有 4 条消息（两问两答）
    assert len(history) == 4
    assert isin

Scaling Rate Limiting from Single‑Node to a Distributed Go+Redis Token Bucket — 10x Throughput Under Load (with Degradation Strategy)

BAOFUFAN — Sun, 03 May 2026 12:08:09 +0000

At 2 AM, an alert pulled me out of bed — the database connection pool of our order service was exhausted, and most requests were returning 504. It turned out a marketing campaign was driving triple the usual traffic. Our in‑memory per‑instance token bucket rate limiter, deployed across three replicas, operated in isolation; global rate limiting was effectively non‑existent. That moment I realized: if the state is not shared, rate limiting is just an illusion.

Breaking Down the Problem

This is distressingly common in microservices. To protect downstream services, teams often set a limit like “max 200 QPS per instance”. Deploy three instances, and you might assume global traffic will be capped at 600 QPS. In reality, load balancing is rarely perfectly even — one instance exhausts its 200 QPS quota while the other two still have headroom. The actual peak hitting the downstream can easily exceed 900 QPS. This is the fatal flaw of per‑instance rate limiting at scale: the limiting logic is chopped up by instance boundaries, becoming “paper‑only” rate limiting from a global perspective.

The root cause is simple: the token bucket’s current token count and last refill timestamp live purely in memory and are not shared across instances. A typical Redis fixed‑window counter (INCR + EXPIRE) can share state, but it suffers from boundary spikes — the last 100 ms of one second and the first 100 ms of the next can overlap to produce a burst of twice the allowed rate, still dangerous for downstream systems. We needed a solution that shares state and smooths traffic — a distributed token bucket.

Design

Choice: Go + Redis + Lua script for a distributed token bucket.

Why not the other options?

Nginx/gateway‑level rate limiting: Adds a proxy hop and decouples from business logic, making it hard to implement fine‑grained controls (e.g., mixed limiting by user and API).
Pure Redis sliding window: Doable with sorted sets, but you must constantly evict expired members, incurring memory and CPU overhead, and the algorithmic complexity often introduces performance bottlenecks.
Go distributed rate‑limiting libraries: Many are unmaintained or only support simple fixed‑window counters, lacking the flexibility of a token bucket.

The final architecture is straightforward: move the token bucket’s core state (tokens, last_refill_time) into Redis, and use a Lua script to atomically calculate and update them. Thanks to Redis’s single‑threaded execution model, unlimited concurrent requests are serialized safely. The application side wraps this in a DistributedTokenBucket struct that integrates a built‑in degradation strategy: when Redis is unavailable (timeout, disconnection), it automatically falls back to a local golang.org/x/time/rate token bucket. Even if Redis completely goes down, downstream services are not overwhelmed — we degrade to single‑instance rate limiting, preserving the fundamental protection.

Core Implementation

The following Lua script handles the atomic “token generation + consumption check” step. It accepts the timestamp as an argument to avoid relying on potentially inconsistent system clocks across instances. (Using redis.call('TIME') is also an option, depending on your consistency paranoia.)

// 这段代码解决：如何用一段 Lua 保证“计算新增令牌 -> 判断是否足够 -> 扣减”的原子性
const tokenBucketLua = `
local key       = KEYS[1]              -- 令牌桶 key
local rate      = tonumber(ARGV[1])    -- 每秒生成令牌数
local capacity  = tonumber(ARGV[2])    -- 桶容量
local now       = tonumber(ARGV[3])    -- 当前时间戳（毫秒）
local requested = tonumber(ARGV[4])    -- 请求令牌数

local bucket = redis.call('HMGET', key, 'tokens', 'last_refill')
local tokens = tonumber(bucket[1])
local last_refill = tonumber(bucket[2])

if tokens == nil then
    -- 首次访问，初始化令牌桶
    tokens = capacity
    last_refill = now
end

-- 计算经过的时间及新增令牌数
local delta = math.max(0, now - last_refill)
local new_tokens = math.floor(delta * rate / 1000)
tokens = math.min(capacity, tokens + new_tokens)

local allowed = 0
if tokens >= requested then
    tokens = tokens - requested
    allowed = 1
end

-- 更新 Redis 中的状态，并设置一个合理的 TTL 防止冷数据残留
redis.call('HMSET', key, 'tokens', tokens, 'last_refill', now)
redis.call('EXPIRE', key, 60)

return {allowed, tokens}
`

Next, the Go struct and the core Take method. Its responsibility is to execute the Lua script, handle Redis errors, and trigger the fallback path when Redis is not healthy.

// 这段代码解决：封装 Redis 调用，提供限流入口，并在 Redis 不可用时降级到本地令牌桶
import (
    "context"
    "errors"
    "time"

    "github.com/redis/go-redis/v9"
    "golang.org/x/time/rate"
)

type DistributedTokenBucket struct {
    rdb        *redis.Client
    script     *redis.Script
    key        string
    rate       float64 // 令牌/秒
    capacity   int     // 桶容量
    fallback   *rate.Limiter // 本地降级限流器
}

func NewDistributedTokenBucket(rdb *redis.Client, key string, ratePerSec float64, capacity int) *DistributedTokenBucket {
    // 本地降级器：容量和速率取全局值的一部分，保护下游
    fallbackLimiter := rate.NewLimiter(rate.Limit(ratePerSec), capacity)
    return &DistributedTokenBucket{
        rdb:      rdb,
        script:   redis.NewScript(tokenBucketLua),
        key:      key,
        rate:     ratePerSec,
        capacity: capacity,
        fallback: fallbackLimiter,
    }
}

func (b *DistributedTokenBucket) Take(ctx context.Context) bool {
    now := time.Now().UnixMilli()
    result, err := b.script.Run(ctx, b.rdb, []string{b.key}, b.rate, b.capacity, now, 1).Result()
    if err != nil {
        // Redis 不可用时，降级为本地令牌桶
        return b.fallback.Allow()
    }

    values, ok := result.([]interface{})
    if !ok || len(values) < 1 {
        return b.fallback.Allow()
    }

    allowed, ok := values[0].(int64)
    if !ok {
        return b.fallback.Allow()
    }

    return allowed == 1
}

This design ensures the happy path is fully distributed and cooperative, while the unhappy path keeps the system alive. No single point of failure in Redis should ever bypass all protection.

In our load tests, replacing the old per‑instance token bucket with this distributed implementation allowed us to safely absorb a 10x increase in global QPS without crashing the downstream. The fallback kicked in seamlessly during Redis failover, proving that the “paper‑only” days were over.

Slash Multi-Level Cache Debugging Time by 90% with Pytest Parametrization

BAOFUFAN — Sun, 03 May 2026 01:08:38 +0000

The winter in Hangzhou is miserably damp. At 1:47 AM, I was jolted awake by an alert SMS — “User profile page returning mixed values, user A is seeing user B’s orders.” My gut told me it was cache corruption again. After digging around for a while, I found that the invalidation logic between the local lru_cache and Redis had missed a single delete in one branch. I had to manually run dozens of test cases just to reproduce it. The next day, I refactored these tests using Pytest parameterization, turning “manual brain exhaustion” into “automated machine exhaustion.” I’ve never lost sleep over this issue since. This article is about how to use Pytest parameterization to achieve zero-blind-spot testing for multi‑level cache (local + Redis) consistency verification.

Why Manually Testing Multi‑Level Caches Is a Bottomless Pit

Multi‑level caching is a common pattern: read requests first check a local store (e.g., lru_cache or cachetools); on a miss, they hit Redis and then backfill the local cache. Writes update Redis and selectively invalidate the local cache. That "selective invalidation" is a hotbed for bugs — you often skip clearing the local cache on certain update paths for performance reasons, and then a path you thought was safe suddenly breaks.

For example, an endpoint that changes a username only deletes the Redis key user:{id}, but the local cache key happens to be user_profile:{id}. That’s a miss. More subtly, the local TTL is very short. During peak hours, high QPS constantly rebuilds the cache, masking the inconsistency; it’s only exposed late at night when traffic drops. Behavior in the test environment and production look completely different.

Typical manual testing needs to cover: multi‑key mappings, reads after concurrent updates, backfill on cache miss, TTL expiration boundaries, in‑process mutual exclusion, and more. A human brain can enumerate maybe 20 combinations and still often falls short. Pytest parametrization automates this entire process, and test cases double as documentation, so even newcomers understand them in seconds.

Design: Use `@pytest.mark.parametrize` to Build a “Scenario Matrix”

My goal wasn’t to test the caching middleware itself but to verify that the business logic’s composition is correct. So I adopted a layered testing approach:

Fake Redis (using the fakeredis library) to eliminate external dependencies and let tests run directly in CI.
The system under test is a CacheManager class that encapsulates the strategy: “local read → Redis read → local backfill” as well as “write Redis + local cleanup.”
Test cases are generated via parameterization, covering: whether a key hits in local store, whether it hits Redis, whether backfill occurs, whether the local cache is correctly deleted after a write, and whether dirty reads happen under concurrent access.

Why not use integration tests against a real Redis? Speed. These parameterized cases will eventually cover hundreds of combinations; a unit test must complete in milliseconds, otherwise nobody will run them frequently. And no Docker dependency means what‑you‑see‑is‑what‑you‑get.

Core Implementation: Multi‑Level Cache Class + Pytest Parameterized Tests

1. The `CacheManager` under test (ready to run)

This code implements the read path (“local first, then remote”) and the write path (“remote first, then clear local”).

# cache_manager.py
import time
from functools import lru_cache
import redis as redis_lib

class CacheManager:
    """本地(LRU) + Redis 两级缓存管理器"""
    def __init__(self, redis_client: redis_lib.Redis, local_ttl: int = 60):
        self.redis = redis_client
        self.local_ttl = local_ttl
        # 本地缓存，最多存 128 个 key，用于实际业务限制内存
        self._local_store = {}

    def _local_get(self, key: str):
        """从本地字典读，并检查过期时间"""
        entry = self._local_store.get(key)
        if not entry:
            return None
        if time.time() - entry["ts"] > self.local_ttl:
            del self._local_store[key]
            return None
        return entry["value"]

    def _local_set(self, key: str, value: str):
        self._local_store[key] = {"value": value, "ts": time.time()}

    def _local_delete(self, key: str):
        self._local_store.pop(key, None)

    def get(self, key: str) -> str | None:
        # 1. 先查本地
        val = self._local_get(key)
        if val is not None:
            return val

        # 2. 再查 Redis
        val = self.redis.get(key)
        if val is not None:
            # 3. 回填本地缓存，注意解码
            decoded = val.decode() if isinstance(val, bytes) else val
            self._local_set(key, decoded)
            return decoded
        return None

    def set(self, key: str, value: str, ttl: int = 300):
        # 先写远程，再清本地，保证下次本地读强一致
        self.redis.setex(key, ttl, value)
        # 这里故意只清本地，依赖下次 get 回填
        self._local_delete(key)

2. Pytest parameterized tests – covering read‑write combinations

The code below solves the problem of exhaustively iterating all permutations: “local hit/miss × Redis hit/miss × read‑after‑write,” verifying both the correctness of the returned values and the backfill logic.

# test_cache_consistency.py
import pytest
import redis as redis_lib
from fakeredis import FakeRedis
from cache_manager import CacheManager

@pytest.fixture
def fake_redis():

From 800 Lines of Shell to 30 Lines of Pytest: 10x Redis Persistence Testing Efficiency

BAOFUFAN — Sat, 02 May 2026 12:08:06 +0000

It was 2 a.m. when I got jolted awake by an alerting call—all user points data had rolled back by three hours. After digging for ages, I found that ops had tweaked the save parameter in redis.conf, changing the RDB snapshot interval from 5 minutes to 3 hours. When the node restarted, a massive amount of hot data simply evaporated. What made it worse: this configuration change had been “tested manually”. A colleague restarted Redis, saw that the keys were still there, and called it good. I cursed at the screen: “What’s the point of testing if you test like this?”

The next day, I tore down the entire persistence verification setup and rebuilt it with pytest + Docker as an automated test suite. What used to take 800 lines of Shell and 2 hours of environment tweaking now runs in a few minutes with 30 lines of pytest. Best of all, any reckless change to the persistence configuration can be proven within 10 seconds—did we lose data or not?

Breaking down the problem: why manual Shell/Docker persistence tests are basically useless

Redis persistence comes in three flavors: RDB, AOF, and a mix of both, plus a jungle of parameters like save, appendfsync, aof-use-rdb-preamble, and many more—combinatorial explosion. Most teams verify persistence in one of two ways:

Manually starting and stopping Docker containers, writing a few items with redis-cli, doing docker restart, then running KEYS *—which only proves “it can start”, not “how many seconds of data disappeared”.
Writing a pile of Shell scripts that use docker exec to drive redis-cli and then diff the data—scripts that get bloated and are brittle because the environment changes every time: docker stop wait time, file cleanup policies—even a minor change makes results unpredictable.

The root cause is clear: Redis persistence is the product of a time window, system signals, and filesystem flushing. Manual operation simply can’t control these precisely. For example, docker stop sends SIGTERM to the container by default; when Redis receives it, it tries to perform an RDB save. But how long does that save take? Will it be cut off by SIGKILL? A Shell script has no ability to simulate fault scenarios like “how much data is lost at the moment of a crash.” Even more importantly, consistency verification lacks repeatable assertions—manual testing only gives you a gut feeling that “probably nothing was lost.” That’s a landmine for production.

Solution design: why pytest + Docker, not Testcontainers or a K8s Job?

I wanted a programmable, assertable, reproducible test framework with these core requirements:

Precisely control Redis startup parameters and persistence configuration
Simulate real-world failures: kill -9, power-off-style shutdown, AOF file truncation, etc.
Automatically clean up the environment after a run—no leftover garbage
Run in CI/CD, but also instantly on a dev machine

Technology comparison:

Solution	Pros	Why not chosen
Shell + docker-compose	Team familiarity	Weak assertions, unable to precisely control restarts and signals, shell script maintenance nightmare
Testcontainers (Python)	Native pytest integration, good lifecycle management	Can only operate parameters through `redis-cli` after initialization? In reality, config changes (e.g., dynamic AOF switching) require another wrapper; also the underlying `docker-java` isn’t very friendly to Python, high debugging cost
Kubernetes Job	Production-grade	Too heavy, can’t run locally, CI needs a K8s cluster – using a sledgehammer to crack a nut
docker-py + pytest	Lightweight, programmable container lifecycle control, native Python assertions	This is the one I chose. Use the `docker` SDK to start/stop containers and manage volumes, `redis-py` for data read/write, pytest fixtures for environment injection. The whole solution is under 500 lines of Python, and on CI it only depends on a Docker daemon.

Architecturally, I split the tests into three layers:

Infrastructure layer: docker-py creates Redis containers, mounts temporary volumes for RDB/AOF files.
Operation layer: redis-py writes, reads, issues CONFIG SET, BGSAVE, etc.
Assertion layer: pytest asserts whether data exists, whether files were created, whether the AOF contains the last write.

This layering lets test cases focus only on “write data → how it dies → is the data correct after restart,” without caring about how the container starts or what mount paths are used.

Core implementation: ready-to-run test code

The following code addresses one problem: verify that after a Redis process is killed with kill -9, all data written after the last BGSAVE is lost as expected—and no extra loss occurs.

1. conftest.py: managing the Redis container lifecycle with a fixture


python
# conftest.py
import pytest
import docker
import redis
import time
import os

REDIS_IMAGE = "redis:7.2"  # 固定版本，避免 CI 上拉取 latest 导致不一致

@pytest.fixture(scope="function")
def rdb_container(tmp_path):
    """
    启动一个配置了 RDB 持久化的 Redis 容器，数据文件写入临时目录。
    tmp_path 是 pytest 提供的临时路径，每个测试函数独立，互不干扰。
    """
    client = docker.from_env()
    data_dir = tmp_path / "data"
    data_dir.mkdir()

    container = client.containers.run(
        image=REDIS_IMAGE,
        name=f"redis-rdb-test-{os.getpid()}",  # 避免容器重名
        command=[
            "redis-server",
            "--save 900 1",        # 900秒内至少1次修改则保存，这里故意设大，手动控制BGSAVE
            "--save 300 10",
            "--save 60 10000",
            "--dir /data",
            "--dbfilename dump.rdb"
        ],
        volumes={str(data_dir): {"bind": "/data", "mode": "rw"}},
        ports={"6379/tcp": None},  # 让 Docker 分配随机端口
        detach=True,
        remov

Uncovering 8% IndexedDB Data Loss After Browser Crashes with Playwright

BAOFUFAN — Sat, 02 May 2026 01:07:53 +0000

At 2 a.m., our user group exploded — people were saying data had just vanished, as if the browser had “eaten” it. Our frontend stores application state in IndexedDB, which is supposed to be far more reliable than localStorage. How could it disappear without a trace? I spent two hours digging through logs and backend records before zeroing in on a dark secret of browser storage: when disk space gets tight, Chrome will silently delete IndexedDB data without any notification. Worse, you can’t reproduce it by hand because you’re not running on the “chosen” hard drive. I decided to write an automated test with Playwright that simulates browser crashes and storage pressure — and expose IndexedDB’s real behavior.

Breaking down the problem

IndexedDB was designed to be a client-side persistent storage, and the W3C spec even says “data should be kept as long as possible”. But a spec is one thing; what browser vendors actually implement is another. Chrome has a mechanism called “Storage Pressure Eviction”: when the user’s disk space drops below a certain threshold, the browser evicts data from less “important” origins using an LRU policy. By default, IndexedDB does not request a persistent-storage permission (navigator.storage.persist()), so it’s very easy to get kicked out. If you haven’t applied for persistent storage permission in a PWA, your database is about as sturdy as a camping tent.

Why don’t normal testing approaches work? Because manual testing only covers “normal reads and writes” — it can’t simulate:

A sudden browser process crash (kill, power loss)
The context being unexpectedly destroyed and then restarted (user closing a tab and reopening it)
The internal cleanup triggered by a disk-space warning

These scenarios require a controlled environment where you can repeatedly run a fast write → destroy → rebuild → verify loop automatically. That’s exactly what Playwright’s Browser Context isolation and its rich CDP (Chrome DevTools Protocol) capabilities are built for.

Solution design

I didn’t choose Selenium because it’s too heavy and context management feels unnatural. I skipped Puppeteer because Playwright natively supports multiple browsers and multiple contexts with a more modern API. Most importantly, each context created by Playwright’s browser.new_context() has its own independent storage sandbox — closing that context is equivalent to destroying the entire session’s IndexedDB, perfectly simulating the “user closes browser / tab” action.

The architecture is a straightforward “brutal loop validation”:

Use Playwright to create a persistent context (so it won’t be automatically cleaned up).
Open the page and inject a script that writes a record with a unique ID and a checksum into IndexedDB, then explicitly call navigator.storage.persist() to request persistence.
Actively close that context to simulate a browser close or crash.
Create a new context, open the same page, read from IndexedDB, and check both data integrity and the number of records.
Repeat N times, each time writing data of random sizes and occasionally using CDP commands to simulate storage-pressure events.
Count the number of data-loss events and inconsistencies, then generate a report.

Why not use incognito mode for this? Because IndexedDB in incognito is designed to be wiped on close — testing persistence there would be pure performance art.

Core implementation

First, install Playwright and pytest. Then you can run the following three pieces of code directly.

Code 1: IndexedDB utility functions — solving “how to reliably write and make sure it’s actually flushed to disk”

This is the foundation. Inside page.evaluate() we wrap the entire IndexedDB transaction lifecycle in a Promise, ensuring the data is committed before returning.

# idb_helpers.py
from playwright.sync_api import Page

IDB_WRITE_SCRIPT = """
async (dbName, storeName, key, value) => {
    return new Promise((resolve, reject) => {
        const request = indexedDB.open(dbName, 1);
        request.onupgradeneeded = (event) => {
            const db = event.target.result;
            if (!db.objectStoreNames.contains(storeName)) {
                db.createObjectStore(storeName, { keyPath: 'id' });
            }
        };
        request.onsuccess = (event) => {
            const db = event.target.result;
            // The transaction scope must include storeName, otherwise the write won't go through
            const tx = db.transaction(storeName, 'readwrite');
            const store = tx.objectStore(storeName);
            // Store a CRC field inside value to verify consistency later
            store.put({ id: key, data: value, checksum: simpleChecksum(value) });
            tx.oncomplete = () => resolve(true);
            tx.onerror = (e) => reject(e);
        };
        request.onerror = (e) => reject(e);

        function simpleChecksum(str) {
            let hash = 0;
            for (let i = 0; i < str.length; i++) {
                hash = ((hash << 5) - hash) + str.charCodeAt(i);
                hash |= 0; // Convert to 32bit integer
            }
            return hash;
        }
    };
}
"""

def write_indexeddb(page: Page, db_name, store_name, key, value):
    return page.evaluate(IDB_WRITE_SCRIPT, db_name, store_name, key, value)

Why did we add a checksum here? Because we

3 Asyncio Pitfalls That Took Me 3 Hours to Debug and Almost Crashed Production

BAOFUFAN — Fri, 01 May 2026 20:25:25 +0000

Here’s the story: last week my lead asked me to optimize a data aggregation service that calls 20 downstream APIs. The serial version took around 18 seconds — users were ready to throw their keyboards. Obvious IO-bound job, right? I thought I’d slap on asyncio, ship it in half a day, and look like a hero. Instead, I spent three hours falling into every rabbit hole asyncio had to offer, and nearly took down production. This post walks through the three biggest pitfalls I hit and how to write async code that actually works in the real world.

Get Your Concepts Straight First

At its core, asyncio is a single-threaded event loop — a master scheduler that lines up coroutines. When one coroutine is waiting on IO, the loop politely tells it to step aside and runs whichever coroutine is ready instead. You only need two keywords: async def to define a coroutine function, and await to yield control, telling the event loop “I’ll be waiting here, go do something else.”

Most tutorials show you this perfect‑world example:

import asyncio

async def fetch(url):
    await asyncio.sleep(1)  # simulate network IO
    return f"data from {url}"

async def main():
    tasks = [fetch(f"api/{i}") for i in range(5)]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

Clean, elegant, 5 requests in 1 second. But the moment you drop this into a real project, things get messy.

Pitfall 1: `await` Inside a Sync Function — And Boom, Errors

I naively added await fetch() right inside an existing Flask route function. Immediate SyntaxError: 'await' outside async function. Alright, I’ll just change the route to async def. Request comes in — RuntimeError: There is no current event loop in thread 'Thread-1'.

Here’s why: Flask uses a thread pool to handle requests. Each worker thread doesn’t have its own event loop, and you can’t just call asyncio.run() inside a thread that already has a loop running. My view ended up calling asyncio.run(main()) and triggered a cascade of “event loop already running” errors.

What you should do: If you can, switch to an async‑native framework like Quart or FastAPI. If you’re stuck with Flask, create a global event loop at startup and schedule work with loop.run_until_complete(). Or, even simpler: spin up a background asyncio thread and communicate with the web thread via a queue.

Pitfall 2: Blocking Calls Inside a Coroutine — Performance Tanks

Feeling clever, I wrote:

results = await asyncio.gather(*[call_api_blocking(i) for i in range(20)])

Total time? Still ~18 seconds. Logging showed each task finishing one after another, no concurrency at all. The culprit: call_api_blocking used requests.get(), a synchronous blocking call. await is useless here — while the first requests.get sits there, the whole thread is frozen and no other coroutine gets a chance to run.

Asyncio only plays nice with its own async IO primitives. When you have a blocking call, you must ship it to a thread pool with loop.run_in_executor():

async def call_api_async(url):
    loop = asyncio.get_running_loop()
    return await loop.run_in_executor(None, requests.get, url)

Now the blocking happens in a separate thread and the event loop can immediately switch to another coroutine. Later I replaced requests with aiohttp entirely, and performance really took off. The golden rule: async is all-or-nothing. Don’t mix in blocking calls that hijack your thread.

Pitfall 3: Orphaned Tasks — Memory Climbs, Then OOM

After performance looked good, I rolled it out. Two days later, the pod was OOMKilled. Memory kept growing slowly, and the GC wasn’t collecting objects. After digging, I found the culprit. To “flexibly control concurrency” I had written something like this:

tasks = []
for url in urls:
    task = asyncio.create_task(process(url))
    tasks.append(task)
for t in tasks:
    await t

Looks fine, right? But inside process(url) some branches returned early, and a few exceptions weren’t handled properly. This left tasks in PENDING or CANCELLED state while still referenced by the tasks list. Those tasks held onto large response data, so the GC chain was never broken — classic memory leak.

The fix: Use asyncio.TaskGroup (Python 3.11+) to manage lifetimes automatically. If any task fails, all others are cancelled and resources are cleaned up:

async def main():
    async with asyncio.TaskGroup() as tg:
        for url in urls:
            tg.create_task(process(url))

If you’re on an older Python version, be diligent about cancelling pending tasks in a finally block and clearing references.

The Production‑Ready Version

Here’s the core skeleton I ended up with — concurrency controlled via semaphore, a reused aiohttp session, isolated exceptions, and timeouts:

import asyncio
import aiohttp
import time
from typing import List

class AsyncFetcher:
    def __init__(self, concurrency: int = 10, timeout: int = 10):
        self.sem = asyncio.Semaphore(concurrency)  # limit concurrency to avoid hammering downstream
        self.timeout = aiohttp.ClientTimeout(total=timeout)

    async def fetch_one(self, session: aiohttp.ClientSession, url: str) -> dict:
        async with self.sem:
            try:
                async with session.get(url, timeout=self.timeout) as resp:
                    data = await resp.json()

I Rewrote Our Crawler with asyncio and Got a 15x Performance Boost

BAOFUFAN — Fri, 01 May 2026 02:00:40 +0000

Last week, I finally snapped. Our “legacy” news aggregator was crawling 200 sites in 8 minutes, with two database timeouts along the way. Ops complained it was “slower than a tortoise,” the product manager asked, “Can we get it under 1 minute?” I said: give me half a day, and I’ll rewrite it with asyncio.

The result? Total time dropped from 487 seconds to 32 seconds — a 15x speedup. My boss walked past my desk, glanced at the screen, and literally said, “Whoa, now that’s the speed it should be.” Today I’ll walk you through that refactor — no textbook fluff, just real, battle‑tested takeaways.

Why asyncio, not threading?

When faced with I/O‑bound tasks, many folks reach for concurrent.futures and thread pools. But threads come with GIL overhead, context‑switching costs, and let’s be honest — a crawler spends 99% of its time waiting for network responses. Using OS threads to “wait for I/O” is like hiring a fleet of drivers just to have them sit in their cars.

asyncio takes a different approach: single thread + event loop. When a coroutine is waiting for a network response, it voluntarily yields control (await), and the event loop immediately switches to another coroutine that’s ready to run. No thread‑switching overhead, no lock contention, minimal memory footprint.

Three core ingredients:

Event loop – the scheduler; it runs whatever is ready.
Coroutines – async def functions that suspend with await.
Futures/Tasks – wrappers around coroutines that let you wait for results.

It’s a completely different mindset from synchronous code — you have to get comfortable thinking concurrently.

The refactor: from blocking sync to async concurrency

Let’s start with the synchronous crawler I inherited (simplified core logic):

import time
import requests

URLS = [f"https://httpbin.org/delay/1?id={i}" for i in range(10)]

def fetch_sync(url: str) -> str:
    # each request blocks for 1 second (simulating network I/O)
    resp = requests.get(url, timeout=5)
    return resp.json()["url"]

start = time.perf_counter()
results = [fetch_sync(url) for url in URLS]
elapsed = time.perf_counter() - start
print(f"同步耗时: {elapsed:.2f}s, 结果数: {len(results)}")
# Output: 同步耗时: 10.12s, 结果数: 10

Ten requests, each taking 1 second, executed one after another — naturally that’s 10 seconds. Who can put up with that?

Converting to asyncio boils down to two steps: swap the I/O function for its async counterpart, then schedule everything concurrently.

import asyncio
import aiohttp
import time

URLS = [f"https://httpbin.org/delay/1?id={i}" for i in range(10)]

async def fetch_async(session: aiohttp.ClientSession, url: str) -> str:
    # aiohttp async request — await yields control
    async with session.get(url, timeout=5) as resp:
        data = await resp.json()
        return data["url"]

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_async(session, url) for url in URLS]
        results = await asyncio.gather(*tasks)   # run all coroutines concurrently
        return results

start = time.perf_counter()
results = asyncio.run(main())
elapsed = time.perf_counter() - start
print(f"异步耗时: {elapsed:.2f}s, 结果数: {len(results)}")
# Output: 异步耗时: 1.05s, 结果数: 10

asyncio.gather() fires off all 10 coroutines at once, so the total time is roughly that of the slowest request (1 second) instead of the sum. That’s the magic of the event loop: while coroutine 1 is waiting on I/O, the loop is already running coroutine 2, 3, … until a response arrives and control is handed back.

Going deeper: semaphores and error handling — don’t let async become chaos

If you think the snippet above is production‑ready, you’re probably in for a rude awakening. The first pitfall I hit was unlimited concurrency. When the URL list grew from 10 to 2,000, the target server instantly banned my IP — because I had opened 2,000 TCP connections at once.

The fix: asyncio.Semaphore, to cap the number of simultaneous coroutines.

async def fetch_with_limit(session, url, sem, retries=3):
    async with sem:  # semaphore controls how many coroutines run at once
        for attempt in range(retries):
            try:
                async with session.get(url, timeout=5) as resp:
                    if resp.status == 200:
                        return await resp.json()
                    else:
                        raise Exception(f"HTTP {resp.status}")
            except Exception as e:
                if attempt == retries - 1:
                    print(f"请求失败: {url}, 错误: {e}")
                    return None
                await asyncio.sleep(2 ** attempt)  # exponential backoff

async def main_with_limit():
    sem = asyncio.Semaphore(50)  # max 50 concurrent requests
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, url, sem) fo

asyncio Pitfalls: The Mistake That Cost Me 3 Hours

BAOFUFAN — Fri, 01 May 2026 01:58:14 +0000

Here’s the story: last week my boss threw a “simple” task at me — pull data from 120 internal APIs simultaneously and compile a report. I thought, “This is just I/O-bound work. I know asyncio like the back of my hand.” So I cranked out the first version in 10 minutes. To my disbelief, it ran even slower than a serial approach, and some endpoints never returned any data. That afternoon, I stared at the terminal output, tweaking and cursing for three full hours — until I spotted one innocuous function call. Then it all clicked.

If you’re doing concurrency with asyncio, the following pitfalls might make you question your life choices.

1. The Culprit: Synchronous Blocking Call Inside a Coroutine

Here’s my first naive implementation — can you spot the problem right away?

import asyncio
import time
import requests  # 注意：经典的同步库

async def fetch_api(url):
    """协程函数：获取 API 数据"""
    print(f"Starting {url}")
    # 模拟获取数据 —— 这里埋了一颗大雷
    resp = requests.get(url, timeout=10)  # 同步阻塞调用！
    data = resp.json()
    print(f"Finished {url}")
    return data

async def main():
    urls = [f"https://httpbin.org/delay/1?req={i}" for i in range(10)]
    start = time.time()
    results = await asyncio.gather(*[fetch_api(u) for u in urls])
    elapsed = time.time() - start
    print(f"10 请求耗时: {elapsed:.2f}s")
    return results

asyncio.run(main())

The result left me dumbfounded: 10 requests took over 10 seconds — exactly like a serial run. The reason is painfully simple: requests.get() is a synchronous blocking call. While waiting for the network, it completely holds the thread hostage, so the event loop never gets a chance to switch to another coroutine. Mixing synchronous code into an async def is like stuffing a tractor engine into a sports car. The golden rule of asyncio is: every I/O operation must be asynchronous.

Two ways to fix it: swap to an async HTTP library (like aiohttp), or offload the blocking call with loop.run_in_executor. I recommend the former:

import aiohttp
import asyncio
import time

async def fetch_api(session, url):
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
        return await resp.json()

async def main():
    urls = [f"https://httpbin.org/delay/1?req={i}" for i in range(10)]
    start = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_api(session, u) for u in urls]
        results = await asyncio.gather(*tasks)
    elapsed = time.time() - start
    print(f"10 请求耗时: {elapsed:.2f}s")
    return results

asyncio.run(main())

After the switch, 10 requests finished in about 1.5 seconds. My boss’s frown finally relaxed.

2. Forgetting to `await` — The Coroutine That Never Ran

This trap has bitten me more times than I’d like to admit. Check out this classic:

async def say_hello():
    await asyncio.sleep(1)
    print("Hello")

async def main():
    # 事故现场：创建协程对象，但忘了 await
    say_hello()          # 只会返回一个 coroutine object，不会执行
    await asyncio.sleep(2)
    print("End")

When you run it, the terminal only prints End. The Hello never appears. Python doesn’t raise an error — it silently creates a coroutine object and drops it into the void. The correct approach is await say_hello(), or wrap it with asyncio.create_task(say_hello()) so the event loop manages it. My personal habit: whenever I call an async def function, I either put await in front of it or wrap it with create_task. I never leave a coroutine naked.

3. Exception Handling in `gather` — One Rotten Task Spoils the Whole Bunch

When I took on that 120‑endpoint task, a few APIs occasionally timed out or returned 500. I used asyncio.gather and quickly learned that if a single task raises an exception, all the other tasks — finished or unfinished — get cancelled, leaving me with zero usable data.

# 错误示范：一个炸，全家炸
async def bad_request():
    await asyncio.sleep(0.5)
    raise ValueError("接口挂了")

async def good_request():
    await asyncio.sleep(1)
    return "正常数据"

async def main():
    try:
        results = await asyncio.gather(bad_request(), good_request())
    except ValueError:
        print("捕获异常，但 good_request 的结果也丢了")

The fix is simple — add return_exceptions=True to gather:

results = await asyncio.gather(
    task1, task2, ...,
    return_exceptions=True
)
for r in results:
    if isinstance(r, Exception):
        log_error(r)      # 单独处理异常
    else:
        process(r)

With this pattern, you can gracefully handle partial failures — log the errors and still process all the valid responses. No more wasting 3 hours staring at the terminal!

These pitfalls are sneaky, but once you understand the underlying mechanics, asyncio becomes a powerful ally. Hope this saves you from the same debugging rabbit hole I fell into.

Forem: BAOFUFAN

Building RAG with LangChain & Chroma: Two Hidden Pitfalls That Cost Me 6 Hours

Why you can’t just dump documents into GPT

Design choices: why Chroma over Pinecone or FAISS

Core implementation: two scripts to run the full RAG pipeline

How I Cut LLM Memory Bug Diagnosis from 2 Hours to 5 Minutes with Playwright & Allure

Problem Breakdown: LLM "Amnesia" Is Scarier Than Hallucination

Solution Design: Why Playwright + Allure?

Core Implementation: From Opening the Browser to Automated Memory Assertions

A 2 AM Serialization Bug in LangChain Memory — And How pytest Stopped It Forever

The Breakdown

The Plan

Core Implementation

1. conftest: Hijacking LangChain’s Redis connection with FakeRedis

2. Testing integrity: write → reload must match bit for bit

"How a Refresh Wiped Out 237 Drafts — and How We Used Playwright to Stop It Forever"

Breaking Down the Problem: Why Memory Storage Fails Silently

Designing the Solution: Why Playwright for Memory Storage Testing

Core Implementation: A Full-Body Checkup for Memory Storage

1. Simulated Page Logic

2. Playwright Test: Verifying Drafts Survive a Refresh

I Spent 6 Hours Fixing LangChain's ConversationBufferMemory — Here's the Automated Test You Need

Breaking Down the Problem

Designing the Solution

Core Implementation

1. Building a Zero-Dependency Test Harness

2. Testing Accuracy: Every Message Written Must Come Back

Scaling Rate Limiting from Single‑Node to a Distributed Go+Redis Token Bucket — 10x Throughput Under Load (with Degradation Strategy)

Breaking Down the Problem

Design

Core Implementation

Slash Multi-Level Cache Debugging Time by 90% with Pytest Parametrization

Why Manually Testing Multi‑Level Caches Is a Bottomless Pit

Design: Use @pytest.mark.parametrize to Build a “Scenario Matrix”

Core Implementation: Multi‑Level Cache Class + Pytest Parameterized Tests

1. The CacheManager under test (ready to run)

2. Pytest parameterized tests – covering read‑write combinations

From 800 Lines of Shell to 30 Lines of Pytest: 10x Redis Persistence Testing Efficiency

Breaking down the problem: why manual Shell/Docker persistence tests are basically useless

Solution design: why pytest + Docker, not Testcontainers or a K8s Job?

Core implementation: ready-to-run test code

1. conftest.py: managing the Redis container lifecycle with a fixture

Uncovering 8% IndexedDB Data Loss After Browser Crashes with Playwright

Breaking down the problem

Solution design

Core implementation

3 Asyncio Pitfalls That Took Me 3 Hours to Debug and Almost Crashed Production

Get Your Concepts Straight First

Pitfall 1: await Inside a Sync Function — And Boom, Errors

Pitfall 2: Blocking Calls Inside a Coroutine — Performance Tanks

Pitfall 3: Orphaned Tasks — Memory Climbs, Then OOM

The Production‑Ready Version

I Rewrote Our Crawler with asyncio and Got a 15x Performance Boost

Why asyncio, not threading?

The refactor: from blocking sync to async concurrency

Going deeper: semaphores and error handling — don’t let async become chaos

asyncio Pitfalls: The Mistake That Cost Me 3 Hours

1. The Culprit: Synchronous Blocking Call Inside a Coroutine

2. Forgetting to await — The Coroutine That Never Ran

3. Exception Handling in gather — One Rotten Task Spoils the Whole Bunch

Design: Use `@pytest.mark.parametrize` to Build a “Scenario Matrix”

1. The `CacheManager` under test (ready to run)

Pitfall 1: `await` Inside a Sync Function — And Boom, Errors

2. Forgetting to `await` — The Coroutine That Never Ran

3. Exception Handling in `gather` — One Rotten Task Spoils the Whole Bunch