1. 项目概述为什么一个LLM SaaS的起点必须是“能登录、能传文件”的骨架做LLM应用开发的朋友大概率都踩过这个坑兴致勃勃搭好大模型推理接口本地调通了连上前端也跑起来了结果一上线就卡在第一步——用户从哪来文件往哪存权限怎么管我去年带团队落地三个垂直领域LLM SaaS产品前两个项目都在“认证上传”环节反复返工第一个项目用自研JWT逻辑两周后发现刷新令牌没做滑动窗口被客户投诉会话频繁中断第二个项目直接套用Django Allauth结果整个用户表结构和OAuth2流程跟FastAPI生态不兼容硬改了四天SQL迁移脚本。直到第三个项目我们才真正把“Auth and File Upload”拆出来作为独立可复用、可审计、可压测的模板层。它不是功能模块而是SaaS系统的呼吸系统——没有它再炫的推理链路也只是离线Demo。这个模板解决的不是“能不能做”而是“能不能稳、能不能审、能不能扩”。它面向三类人刚从Notebook转向生产环境的算法工程师需要零配置接入企业级身份体系独立开发者想快速验证PMFProduct-Market Fit但又不想在登录页上花三天还有技术负责人要确保上传路径符合GDPR/等保2.0对文件元数据、访问日志、生命周期的强制要求。核心关键词——FastAPI、LLM SaaS、Auth、File Upload——不是并列关系而是因果链Auth是文件上传的前提文件上传是LLM服务的输入入口而FastAPI是让这条链路在高并发下不掉链子的底层承重墙。2. 整体设计思路为什么放弃“全栈框架”选择“协议级解耦”2.1 不选Django或Flask的底层逻辑很多人第一反应是“直接用Django Admin配个用户模型再加个django-storages不就完事了”我试过。在内部POC阶段Django确实省了两天时间但当客户提出“必须对接我们AD域的LDAP并且所有登录失败要实时推送到SIEM平台”时问题来了Django的认证中间件是单点强耦合的你得重写authenticate()方法还要绕过它的Session机制去注入自定义审计钩子。更麻烦的是文件上传——Django默认走FileSystemStorage但客户要求所有上传文件必须加密落盘且密钥由HashiCorp Vault动态分发。这时候你会发现框架自带的save()方法根本没法插手加密流。而FastAPI的哲学是“协议即契约”它不预设你用什么数据库、什么存储、什么认证协议只强制你遵守OpenAPI规范和ASGI生命周期。我们把Auth拆成三个协议层传输层强制HTTPS HSTS头所有密码字段用SecretStr封装避免日志泄露协议层OAuth2 Password Flow Bearer Token拒绝任何Cookie-based Session为后续JWT无状态扩展留口实现层用passlib做bcrypt哈希不是argon2因为客户审计要求明确写明“需支持FIPS 140-2验证的哈希算法”而bcrypt有NIST认证版本文件上传同理。我们不用UploadFile直接存磁盘而是抽象出FileStorageBackend接口当前实现是S3CompatibleStorage兼容MinIO、Cloudflare R2、阿里云OSS但预留了EncryptedLocalStorage和IPFSBackend的继承点。这种设计让客户说“下周要切到私有IPFS集群”我们只需要重写一个类改两行DI注入不用碰任何路由或业务逻辑。2.2 为什么Auth必须前置到Lifespan事件中FastAPI的lifespan事件常被当成“启动时初始化DB连接”的地方但我们把它用成了Auth系统的“心跳检测中枢”。模板里有段关键代码app.on_event(startup) async def startup_event(): # 1. 预热JWT密钥缓存避免首请求解密延迟 await jwt_manager.preload_keys() # 2. 检查S3桶策略是否允许POST防止部署后上传500 await file_storage.validate_write_permission() # 3. 启动后台任务每5分钟扫描过期临时上传文件 asyncio.create_task(cleanup_expired_uploads())这段逻辑解决了真实生产中的三个痛点首屏加载慢JWT解密需要RSA公钥如果每次请求都从磁盘读取PEM文件P95延迟会突增80ms。我们提前加载进内存并用cachetools.TTLCache缓存72小时配置漂移S3桶策略可能被运维手动修改导致上传接口静默失败。validate_write_permission()会尝试发一个预签名POST请求失败则直接sys.exit(1)让K8s健康检查立刻拉起新Pod磁盘爆满用户上传10GB文件后关掉页面临时文件不会自动清理。我们用aiocron定时扫描/tmp/uploads/下超过2小时未被file_upload_session表引用的文件——注意这里不是删文件而是先写入cleanup_log表含文件名、SHA256、操作人ID再执行删除满足审计溯源要求。这种把Auth的“可用性保障”下沉到启动阶段的设计让整个SaaS系统从第一行日志开始就处于可审计、可监控状态而不是等用户报错才被动响应。2.3 文件上传的“双通道”架构为什么不用单一multipart/form-dataLLM SaaS的文件上传有两类典型场景小文件高频用户上传PDF合同5MB每秒3-5次要求首字节响应200ms大文件低频用户上传10GB医学影像DICOM序列要求断点续传、分片校验、上传进度可查如果只用FastAPI原生UploadFile会遇到硬伤UploadFile.file.read()是同步阻塞IO在高并发下会吃光Event Loop线程没有分片能力大文件上传超时后只能重传全部浪费带宽无法获取客户端真实IPNginx反代后只有X-Forwarded-For但UploadFile不暴露headers。我们的解法是“双通道”通道A默认POST /v1/upload接收multipart/form-data适用于10MB文件。底层用anyio.to_thread.run_sync()将shutil.copyfileobj()扔进线程池避免阻塞Event Loop通道B大文件专用POST /v1/upload/init返回预签名URL如https://bucket.s3.amazonaws.com/uploads/{uuid}/part-{n}?X-Amz-Signature...客户端直传S3服务端只收POST /v1/upload/complete触发合并逻辑。关键细节在于两个通道最终都写入同一张file_metadata表字段完全一致id,original_name,size_bytes,sha256,uploader_id,status只是storage_path字段值不同通道A存/tmp/uploads/xxx.pdf通道B存s3://bucket/uploads/xxx/。这样下游LLM服务调用get_file_content(file_id)时完全感知不到差异——它只管查表、按storage_path协议去读彻底解耦上传方式与消费方式。3. 核心细节解析Auth与文件上传的12个实操陷阱与避坑方案3.1 JWT Token的“三重过期”设计为什么不能只靠exp字段很多教程教你在create_access_token()里写expires_deltatimedelta(hours1)这在SaaS里是危险操作。我们实际采用“三重过期”机制过期类型字段位置触发条件处理方式Token级过期JWT payloadexp时间到达前端自动跳转登录页Session级过期Redis键session:{user_id}:{jti}用户主动登出/管理员踢出DELETE该键下次请求verify_token()时因jti不存在而拒绝Refresh级过期JWT payloadrefresh_exp刷新令牌超时通常7天强制重新登录提示jtiJWT ID不是UUID而是sha256(f{user_id}_{timestamp}_{random_str})避免Redis键名被枚举。我们用redis-py的pipeline批量执行SET session:{uid}:{jti} 1 EX 604800和SET user:{uid}:current_jti {jti} EX 604800保证原子性。实操中最大的坑是“时钟漂移”。客户服务器时间比NTP源快3分钟导致Token明明没过期却被拒。解决方案在verify_token()里加入leeway120参数python-jose库支持允许最多2分钟偏差。但更根本的是——我们在K8s Deployment里强制挂载hostPath: /etc/timezone并用initContainer运行ntpd -q -p pool.ntp.org校时从基础设施层掐断漂移源头。3.2 密码重置的“时间窗次数锁”双重防护标准的“发送重置邮件→点击链接→填新密码”流程在SaaS里必须防暴力枚举。我们的实现包含三个硬性约束时间窗限制每个邮箱24小时内最多触发3次重置请求。用Redis Hash存储reset_attempts:{email}key为{timestamp}value为1过期时间设为86400秒。每次请求前HLEN reset_attempts:{email}超限返回429 Too Many Requests链接时效性重置Token有效期严格为15分钟且jti绑定用户ID时间戳哈希用完即焚密码强度审计新密码必须通过zxcvbn库检测不是正则要求熵值≥30bit。例如Tr0ub4dour3会被判弱密码因为它是常见单词变形而correct-horse-battery-staple熵值高达44bit直接放行。注意zxcvbn的Python版zxcvbn-python有严重性能问题单次检测耗时200ms我们改用Rust重写的zxcvbn_rs耗时压到8ms内。这是实测下来唯一能在P99延迟50ms要求下可用的方案。3.3 文件上传的“元数据分离”实践为什么不能把文件名存在数据库里新手常犯错误INSERT INTO files (name, size, path) VALUES (?, ?, ?)。这在多语言环境下必崩——用户上传简历_张三.pdfMySQL如果用utf8mb3编码张三会变成??。更糟的是安全风险../../../etc/passwd这种路径遍历如果直接拼接进open()函数就是RCE漏洞。我们的解法是“元数据分离”原始文件名存入original_filename字段类型为VARCHAR(512)字符集utf8mb4安全文件名生成secure_filename f{uuid4().hex}_{int(time.time())}_{hashlib.md5(original_filename.encode()).hexdigest()[:8]}.bin只存这个存储路径固定为/uploads/{year}/{month}/{day}/{secure_filename}由file_storage类统一管理关键代码片段def generate_secure_filename(original: str) - str: # 移除所有控制字符和路径分隔符 clean re.sub(r[\x00-\x1f\x7f/\\:*?|], _, original) # 截断过长文件名Windows限制255字符 if len(clean) 200: name, ext os.path.splitext(clean) clean f{name[:180]}{ext} # 添加时间戳和哈希防重名 timestamp int(time.time()) hash_part hashlib.md5(original.encode()).hexdigest()[:8] return f{uuid4().hex}_{timestamp}_{hash_part}{os.path.splitext(clean)[1] or .bin}实测效果上传../../etc/shadow→ 存为a1b2c3d4_1712345678_1a2b3c4d.bin原始名在数据库里完整保留但物理路径绝对安全。3.4 OAuth2第三方登录的“域白名单”强制校验客户要求“只允许公司邮箱登录”但OAuth2 Provider如Google返回的email字段可被伪造。我们的防护链有三层Provider端配置在Google Cloud Console里设置Authorized domains为yourcompany.com这是最外层防火墙Token校验层verify_google_token()函数不仅验签名还检查hdhosted domainclaim是否存在且等于yourcompany.com数据库约束层users.email字段加唯一索引但类型为VARCHAR(254) COLLATE utf8mb4_0900_as_cs大小写敏感防止ADMINYOURCOMPANY.COM和adminyourcompany.com被当成两个用户。实操心得Google的hdclaim在使用G Suite账号时才返回普通gmail账号没有。所以必须在OAuth2 Flow里强制添加hdyourcompany.com参数否则state参数会被忽略。我们用fastapi-users的OAuth2AuthorizeParams类封装此逻辑避免每个Provider重复写。3.5 大文件上传的“分片校验”实现如何避免网络抖动导致数据损坏当客户端直传S3时分片上传Multipart Upload是标配但默认的ETag校验不可靠——S3对每个Part返回的ETag是MD5哈希但对合并后的ObjectETag可能是abc123...非MD5。我们强制要求客户端在每个Part上传前计算sha256(part_bytes)并在POST /v1/upload/complete时提交所有Part的SHA256列表。服务端收到后调用S3.list_parts()获取所有已上传Part的ETag对每个Part发起HEAD请求读取自定义Headerx-amz-meta-sha256客户端上传时设置比对客户端提交的SHA256与S3返回的x-amz-meta-sha256是否一致全部一致才执行complete_multipart_upload()否则返回400 Bad Request并提示具体哪个Part校验失败。这个设计让网络抖动导致的Part损坏能被精准定位而不是等到LLM服务读取时才发现UnicodeDecodeError。实测中某客户CDN节点故障导致3%的Part上传损坏此机制让问题在10秒内被发现并重传而非等待2小时后用户反馈“文档解析失败”。3.6 Auth中间件的“无感降级”策略当Redis宕机时怎么办Redis是Auth的核心依赖但如果它挂了整个SaaS不能瘫痪。我们的中间件auth_dependency实现了三级降级async def auth_dependency( request: Request, token: str Depends(oauth2_scheme), ) - User: try: # Level 1: 正常走Redis校验jti payload jwt_manager.decode(token) if not await redis_client.exists(fsession:{payload[sub]}:{payload[jti]}): raise HTTPException(401, Invalid session) return await get_user_by_id(payload[sub]) except RedisConnectionError: # Level 2: 降级为JWT签名exp校验无状态 payload jwt_manager.decode_no_redis(token) return await get_user_by_id(payload[sub]) except Exception as e: # Level 3: 最终兜底——记录告警但放行仅限内部测试环境 if settings.ENV staging: logger.warning(fAuth fallback triggered: {e}) return await get_user_by_id(1) # 默认admin raise e关键经验decode_no_redis()函数不查jti但强制校验ississuer、audaudience和exp并用pydantic做payload schema校验。这保证即使Redis全挂用户仍能登录只是无法被管理员踢出——符合“可用性优先于强一致性”的SaaS设计原则。4. 实操过程详解从零搭建可交付的AuthUpload模板4.1 环境准备与依赖锁定为什么用Poetry而不选pip-tools项目根目录的pyproject.toml关键片段[tool.poetry.dependencies] python ^3.11 fastapi {version ^0.110.0, extras [all]} sqlalchemy {version ^2.0.29, extras [mypy]} alembic ^1.13.1 redis ^4.6.0 boto3 ^1.28.50 passlib {version ^1.7.4, extras [bcrypt]} python-jose {version ^3.3.0, extras [cryptography]} zxcvbn-rs ^0.12.0选择Poetry的核心原因是依赖冲突可视化。比如python-jose[cryptography]和cryptography本身有版本交叉约束pip install会静默降级而Poetry的poetry lock --no-update会明确报错“cryptography39.0.0required bypython-jose, butcryptography38.0.4locked”。我们实测过用pip-tools生成的requirements.txt在CI流水线里有7%概率因网络波动导致cryptography安装失败而Poetry的poetry export -f requirements.txt生成的锁文件100%可重现。实操步骤poetry init创建基础配置poetry add fastapi sqlalchemy alembic逐个添加主依赖poetry add --group dev pytest pytest-asyncio httpx添加测试依赖poetry lock生成poetry.lockpoetry export -f requirements.txt --without-hashes requirements.txt供Docker构建使用。特别注意--without-hashes是必须的因为Docker构建时pip install -r requirements.txt如果带hash会因镜像内时区差异导致pydantic校验失败hash计算依赖datetime.now()。4.2 数据库迁移为什么用Alembic而不手写SQLalembic revision --autogenerate -m add users and files tables生成的versions/xxx_add_users_and_files_tables.py内容如下def upgrade(engine): op.create_table( users, sa.Column(id, sa.Integer(), primary_keyTrue), sa.Column(email, sa.String(254), nullableFalse, uniqueTrue), sa.Column(hashed_password, sa.String(128), nullableFalse), sa.Column(is_active, sa.Boolean(), defaultTrue), sa.Column(created_at, sa.DateTime(), defaultsa.func.now()), sa.Column(updated_at, sa.DateTime(), defaultsa.func.now(), onupdatesa.func.now()), ) op.create_table( file_metadata, sa.Column(id, sa.Integer(), primary_keyTrue), sa.Column(original_filename, sa.String(512), nullableFalse), sa.Column(secure_filename, sa.String(255), nullableFalse, uniqueTrue), sa.Column(size_bytes, sa.BigInteger(), nullableFalse), sa.Column(sha256, sa.String(64), nullableFalse), sa.Column(uploader_id, sa.Integer(), sa.ForeignKey(users.id), nullableFalse), sa.Column(status, sa.Enum(FileStatus), defaultFileStatus.UPLOADED), sa.Column(created_at, sa.DateTime(), defaultsa.func.now()), ) # 关键添加索引提升查询性能 op.create_index(ix_file_metadata_uploader_id, file_metadata, [uploader_id]) op.create_index(ix_file_metadata_status, file_metadata, [status])为什么必须加索引实测数据当file_metadata表有50万条记录时SELECT * FROM file_metadata WHERE uploader_id123不加索引耗时2.3秒加索引后压到12ms。而LLM服务每次处理请求都要查用户最近上传的3个文件这个查询是QPS最高的SQL之一。迁移执行命令本地开发alembic upgrade head生产部署在K8s Job里执行alembic upgrade ${MIGRATION_VERSION}用initContainer确保DB连接成功后再运行回滚alembic downgrade -1只允许回退一步强制人工审核。注意alembic revision --autogenerate会漏掉Enum类型变更必须手动在upgrade()里补op.execute(ALTER TYPE file_status ADD VALUE PROCESSING)否则PostgreSQL会报错。4.3 Auth路由实现从/login到/me的完整链路auth/router.py核心代码router.post(/login) async def login( form_data: OAuth2PasswordRequestForm Depends(), db: AsyncSession Depends(get_db), ) - TokenResponse: user await authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException(400, Incorrect email or password) if not user.is_active: raise HTTPException(400, Inactive user) # 生成Access Token和Refresh Token access_token_expires timedelta(hours1) refresh_token_expires timedelta(days7) access_token create_access_token( data{sub: user.id, jti: str(uuid4())}, expires_deltaaccess_token_expires, ) refresh_token create_refresh_token( data{sub: user.id, jti: str(uuid4())}, expires_deltarefresh_token_expires, ) # 将refresh_token jti存入Redis await redis_client.setex( frefresh:{user.id}:{refresh_token_jti}, int(refresh_token_expires.total_seconds()), 1 ) return TokenResponse( access_tokenaccess_token, token_typebearer, expires_inint(access_token_expires.total_seconds()), refresh_tokenrefresh_token, ) router.get(/me, response_modelUserRead) async def read_users_me(current_user: User Depends(current_active_user)): return current_user关键细节OAuth2PasswordRequestForm自动从x-www-form-urlencoded解析username/password无需手动request.form()create_access_token()和create_refresh_token()是两个独立函数因为Refresh Token的scope应更小只含refresh且jti必须不同current_active_user依赖项里get_current_user()先解Token再查DB最后校验is_active三层校验缺一不可实测陷阱OAuth2PasswordRequestForm默认不支持application/json如果前端用fetch()发JSON会返回422 Unprocessable Entity。解决方案是在main.py里加全局异常处理器app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc): if application/json in request.headers.get(content-type, ): # 尝试从JSON body提取username/password try: body await request.json() if username in body and password in body: # 重写form_data逻辑... except: pass return JSONResponse({detail: Invalid request}, status_code422)4.4 文件上传路由支持multipart和pre-signed URL双模式upload/router.py核心实现router.post(/upload, response_modelFileUploadResponse) async def upload_file( file: UploadFile File(...), current_user: User Depends(current_active_user), db: AsyncSession Depends(get_db), file_storage: FileStorageBackend Depends(get_file_storage), ) - FileUploadResponse: # 1. 校验文件大小前端JS校验可被绕过 if file.size settings.MAX_UPLOAD_SIZE: raise HTTPException(413, fFile too large. Max size: {settings.MAX_UPLOAD_SIZE}) # 2. 生成安全文件名 secure_name generate_secure_filename(file.filename) # 3. 保存到存储后端 storage_path await file_storage.save( filefile, secure_filenamesecure_name, user_idcurrent_user.id, ) # 4. 写入数据库 file_meta FileMetadata( original_filenamefile.filename, secure_filenamesecure_name, size_bytesfile.size, sha256await calculate_sha256(file), uploader_idcurrent_user.id, storage_pathstorage_path, ) db.add(file_meta) await db.commit() await db.refresh(file_meta) return FileUploadResponse( idfile_meta.id, original_filenamefile.filename, size_bytesfile.size, uploaded_atfile_meta.created_at, ) router.post(/upload/init, response_modelPreSignedUrlResponse) async def init_multipart_upload( filename: str Form(...), current_user: User Depends(current_active_user), file_storage: FileStorageBackend Depends(get_file_storage), ) - PreSignedUrlResponse: # 生成唯一upload_id upload_id str(uuid4()) # 获取预签名URLS3兼容 presigned_url await file_storage.generate_presigned_post( filenamefilename, upload_idupload_id, user_idcurrent_user.id, ) return PreSignedUrlResponse( upload_idupload_id, presigned_urlpresigned_url, fieldspresigned_url.fields, # 包含x-amz-signature等 )关键参数说明settings.MAX_UPLOAD_SIZE 10 * 1024 * 102410MB这是multipart通道的硬上限generate_presigned_post()返回的fields包含key,x-amz-algorithm,x-amz-credential,x-amz-date,x-amz-signature客户端需用FormData.append()逐个设置upload_id不存数据库只作为S3key前缀如s3://bucket/uploads/{upload_id}/part-1避免元数据膨胀。实测中某客户前端用axios发multipart但没设Content-Type: multipart/form-data; boundaryxxx导致FastAPI解析失败。解决方案在main.py加中间件自动修复app.middleware(http) async def fix_multipart_header(request: Request, call_next): if request.method POST and multipart/form-data in request.headers.get(content-type, ): # 强制设置boundary即使header里没写 if boundary not in request.headers[content-type]: new_boundary f----WebKitFormBoundary{uuid4().hex} request._headers MutableHeaders(request.headers) request._headers[content-type] fmultipart/form-data; boundary{new_boundary} response await call_next(request) return response4.5 测试覆盖为什么用HTTPX而不选TestClienttests/test_auth.py示例pytest.mark.asyncio async def test_login_success(client: AsyncClient, test_user: User): response await client.post( /auth/login, data{username: test_user.email, password: testpassword}, ) assert response.status_code 200 data response.json() assert access_token in data assert data[token_type] bearer # 验证token能解码 payload jwt.decode(data[access_token], settings.JWT_SECRET_KEY, algorithms[HS256]) assert payload[sub] test_user.id pytest.mark.asyncio async def test_upload_file(client: AsyncClient, test_user: User, auth_header: dict): # 创建测试文件 test_file bHello, LLM SaaS! response await client.post( /upload/upload, files{file: (test.txt, test_file, text/plain)}, headersauth_header, ) assert response.status_code 200 data response.json() assert data[original_filename] test.txt assert data[size_bytes] len(test_file)选择httpx.AsyncClient而非TestClient的原因TestClient是同步的无法测试async def路由的真实行为httpx.AsyncClient能真实模拟HTTP/2、流式响应、超时重试等网络行为可以用pytest-asyncio标记与AsyncSession测试无缝集成。实操技巧auth_headerfixture这样写pytest.fixture def auth_header(test_user: User) - dict: token create_access_token(data{sub: test_user.id}) return {Authorization: fBearer {token}}避免在每个test里重复生成token提升测试速度。5. 常见问题与排查技巧实录来自12个生产环境的真实战报5.1 问题速查表高频故障与根因定位现象日志线索根因解决方案401 Unauthorized频发JWT decode error: Signature verification failedNginx代理转发时修改了Authorizationheader大小写如authorization在Nginx配置中加proxy_pass_request_headers on;并确保underscores_in_headers on;文件上传后500 Internal Server ErrorOSError: [Errno 28] No space left on device/tmp分区满Docker容器默认/tmp只有64MB在Dockerfile里加RUN mkdir -p /data/uploads chmod 777 /data/uploadssettings.UPLOAD_DIR /data/uploads422 Unprocessable Entityon/logindetail: [{loc:[body],msg:field required,type:value_error.missing}]前端发JSON但后端期待x-www-form-urlencoded检查前端Content-Type或按4.3节加全局异常处理器403 Forbiddenon S3 pre-signed URLThe request signature we calculated does not match the signature you provided.客户端上传时Content-Type与预签名时声明的不一致在generate_presigned_post()里显式指定Fields{Content-Type: binary/octet-stream}客户端必须匹配504 Gateway Timeouton large file uploadNginx error log:upstream timed out (110: Connection timed out)Nginxproxy_read_timeout默认60秒大文件上传超时在Nginx配置中加proxy_read_timeout 3600;1小时5.2 Redis连接池泄漏的隐蔽征兆与修复某次上线后Redis连接数从20飙升到1024INFO clients显示connected_clients:1024但client list里只有20个活跃连接。根因是redis-py的连接池未正确关闭。我们在lifespan里这样修复app.on_event(shutdown) async def shutdown_event(): # 显式关闭Redis连接池 if hasattr(redis_client, connection_pool): await redis_client.connection_pool.disconnect() # 关闭数据库连接池 await db_engine.dispose()关键经验redis-py的disconnect()必须await否则是异步任务可能被Event Loop丢弃。我们用pytest的monkeypatch测试此逻辑def test_shutdown_disconnects_redis(monkeypatch): mock_pool Mock() mock_pool.disconnect AsyncMock() monkeypatch.setattr(redis_client, connection_pool, mock_pool) # 触发shutdown asyncio.run(shutdown_event()) mock_pool.disconnect.assert_called_once()5.3 PostgreSQL死锁的现场还原与预防死锁日志示例ERROR: deadlock detected DETAIL: Process 12345 waits for ShareLock on transaction 67890; blocked by process 67890. Process 67890 waits for ShareLock on transaction 12345; blocked by process 12345.根因是两个事务以不同顺序更新users和file_metadata表。我们的预防措施统一更新顺序所有事务先UPDATE users再UPDATE file_metadata绝不颠倒减少事务粒度/upload/complete不再在一个事务里完成“更新file_metadata状态插入processing_job”而是拆成两个独立API加锁提示在SELECT ... FOR UPDATE时加SKIP LOCKED避免长事务阻塞。