Supabase多用户并发访问触发RLS 42501错误求助

基于Python(>=3.9) + FastAPI开发，通过Vercel部署，用Supabase实现数据读写。单用户访问正常，多用户并发时，除一名用户外其余均返回RLS权限错误：

{
    "error": "Failed to get history: {'code': '42501', 'details': None, 'hint': None, 'message': 'new row violates row-level security policy for table \"chat_history\"'}"
}

表结构与RLS策略 #

CREATE TABLE chat_history (
  chat_id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
  user_id UUID NOT NULL REFERENCES auth.users(id) ON DELETE CASCADE,
  messages JSONB NOT NULL DEFAULT '[]',
  updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_chat_history_user_id ON chat_history(user_id);


CREATE POLICY "Users can view their own chat history"
ON public.chat_history
FOR SELECT
TO authenticated
USING (
  user_id = auth.uid()
);


CREATE POLICY "Users can insert their own chat history"
ON public.chat_history
FOR INSERT
TO authenticated
WITH CHECK (
  user_id = auth.uid()
);


CREATE POLICY "Users can update their own chat history"
ON public.chat_history
FOR UPDATE
TO authenticated
USING (
  user_id = auth.uid()
)
WITH CHECK (
  user_id = auth.uid()
);


CREATE POLICY "Users can delete their own chat history"
ON public.chat_history
FOR DELETE
TO authenticated
USING (
  user_id = auth.uid()
);

核心Python代码 #

获取/创建聊天记录函数 #

async def _get_or_create_chat_record(self, user_id: str):
    """Get or create a chat history record for the user"""
    # Check if a chat record already exists
    try:
        response = supabase.table("chat_history").select("*").eq("user_id", user_id).execute()
        logger.info(f"Checking chat history for user {user_id}: {response}")
        if response.data and len(response.data) > 0:
            # Chat record exists, return it
            return response.data[0]
        else:
            # Create new chat record
            insert_response = supabase.table("chat_history").insert({
                "user_id": user_id,
                "messages": []  # Empty array of messages
            }).execute()
            
            if not insert_response.data:
                logger.error(f"Failed to create chat history record: {insert_response}")
                raise Exception("Failed to create chat history record")
            
            logger.info(f"Insert response {insert_response.data[0]} for user {user_id}")
            return insert_response.data[0]
    except Exception as e:
        logger.error(f"Error getting or creating chat record for user {user_id}: {str(e)}")
        raise

Auth依赖函数 #

async def get_user_supabase(request: Request):
    """
    Dependency to extract and validate Supabase Auth user from Authorization header.
    """
    auth_header = request.headers.get("Authorization")
    if not auth_header or not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
    access_token = auth_header.split(" ")[1]
    try:
        user_response = supabase.auth.get_user(jwt=access_token)
        if not user_response or not user_response.user:
            raise HTTPException(status_code=401, detail="Invalid or expired token")
        return {
            "supabase": supabase,
            "user": user_response.user
        }
    except Exception as e:
        raise HTTPException(status_code=401, detail=f"Authentication error: {str(e)}")

API路由 #

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest, supabase_client = Depends(get_user_supabase)):

补充信息 #

关闭RLS后多用户并发正常，说明问题出在RLS策略的上下文传递逻辑上。

1. 为每个请求创建独立的Supabase客户端 #

Supabase Python客户端非线程安全，并发请求共享同一客户端时，Auth token会被覆盖，导致RLS验证时auth.uid()拿到其他用户的ID，触发权限错误。修改get_user_supabase函数，为每个验证通过的用户创建独立客户端实例：

from supabase import create_client, Client
import os

# 全局读取Supabase配置（从环境变量获取）
SUPABASE_URL = os.getenv("SUPABASE_URL")
SUPABASE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY")

async def get_user_supabase(request: Request):
    auth_header = request.headers.get("Authorization")
    if not auth_header or not auth_header.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header")
    access_token = auth_header.split(" ")[1]
    
    try:
        # 为当前用户创建独立的Supabase客户端
        user_supabase: Client = create_client(SUPABASE_URL, SUPABASE_KEY)
        # 绑定当前用户的access token
        user_supabase.auth.set_session(access_token)
        
        user_response = user_supabase.auth.get_user()
        if not user_response or not user_response.user:
            raise HTTPException(status_code=401, detail="Invalid or expired token")
        
        return {
            "supabase": user_supabase,
            "user": user_response.user
        }
    except Exception as e:
        raise HTTPException(status_code=401, detail=f"Authentication error: {str(e)}")

2. 修复并发竞态问题 #

原_get_or_create_chat_record的"查询-插入"逻辑存在竞态：并发时多个请求可能同时查到无记录，重复插入导致RLS错误。改用Supabase的upsert原子操作完成逻辑：

async def _get_or_create_chat_record(self, user_id: str, supabase: Client):
    """Get or create a chat history record for the user"""
    try:
        # 原子化upsert操作，避免竞态
        response = supabase.table("chat_history")\
            .upsert({
                "user_id": user_id,
                "messages": []
            }, on_conflict="user_id")\
            .select("*")\
            .execute()
        
        if not response.data:
            logger.error(f"Failed to get or create chat history for user {user_id}: {response}")
            raise Exception("Failed to get or create chat history record")
        
        logger.info(f"Got chat history for user {user_id}: {response.data[0]}")
        return response.data[0]
    except Exception as e:
        logger.error(f"Error getting or creating chat record for user {user_id}: {str(e)}")
        raise

需先给user_id添加唯一约束：

ALTER TABLE chat_history ADD CONSTRAINT unique_user_id UNIQUE (user_id);

3. 验证上下文传递 #

确保每个请求的Supabase客户端都正确绑定当前用户的token，这样auth.uid()才能返回正确的用户ID，匹配RLS策略中的user_id = auth.uid()条件。

内容的提问来源于stack exchange，提问作者Ajin T