[!info] related notes

• 所属 MOC:
• 相关概念:
• 易混淆概念:
• 相关资源:

BodySense KnowledgeLibrary 的 Async Postgres Pool

核心问题

BodySense 旧 KnowledgeLibrary 的公开 API 已经写成：

async def search(...): ...
async def ingest_generated_pack(...): ...
async def list_sources(...): ...
async def stats(...): ...

但内部长期持有：

self._connection: psycopg.Connection | None

并通过同步：

psycopg.connect(...)
with conn.cursor() as cur:
    cur.execute(...)
    rows = cur.fetchall()

完成数据库访问。

因此真正的问题不是“有没有 pool”这么简单，而是四个边界混在一起：

1. Event Loop execution boundary：同步 DB wait 会阻塞 loop。
2. Resource ownership：连接何时创建、何时关闭、谁负责？
3. Concurrency：多个 search 如何安全竞争有限数据库资源？
4. Transaction semantics：多表 ingestion 是否仍然原子？

为什么一个长期 Singleton Connection 不是正确并发模型

旧模型近似：

process
  └─ global KnowledgeLibrary
       └─ one psycopg.Connection

它看起来“省连接”，但隐藏了很多状态。

并发语义模糊

多个 coroutine 同时调用 KnowledgeLibrary.search()，最终都可能触碰同一个长期 connection。

Event Loop 只有一个主调度线程，并不意味着数据库只需要一个 session。多个 coroutine 可以同时处于“等待 DB”的状态，driver/connection 自身还有 transaction 与 wire protocol 约束。

Connection Failure 被扩大成 Global State

长期连接可能：

• server idle timeout；
• network reset；
• transaction aborted state；
• failover；
• database restart。

如果所有请求都绑到它，一个坏 connection 会成为整个 KnowledgeLibrary 的隐式全局故障。

Lifecycle 隐藏在第一位用户请求里

旧 _get_connection() 常常是：

first search arrives
→ lazy connect

这意味着：

application startup successful

并不证明：

database dependency is usable

第一位用户替服务做了依赖探针。

生产系统更适合在 lifespan/readiness 中显式处理长期资源。

为什么“每次请求新建一个 AsyncConnection”也不是理想答案

直觉修复：

async def search(...):
    conn = await AsyncConnection.connect(...)
    ...
    await conn.close()

虽然不再同步阻塞，但每次 interactive search 都重复：

TCP connect
→ PostgreSQL handshake
→ authentication
→ session setup
→ pgvector type registration

这会造成：

• connection setup latency；
• database connection churn；
• burst 时连接风暴。

所以生产更常使用：

少量长期物理连接 + 每次 operation 短期 checkout lease。

Pool 的核心心智：Pool 长期存在，Connection Lease 很短

FastAPI lifespan
  └─ AsyncConnectionPool
       ├─ physical connection 1
       ├─ physical connection 2
       └─ ... bounded

Search A
→ checkout #1
→ query
→ return #1

Search B
→ checkout #2
→ query
→ return #2

因此需要区分三个 owner：

Pool ownership
= application lifespan

Connection ownership
= one short operation / lease

Transaction ownership
= one atomic domain operation

这是 L4 非常重要的 resource-lifetime 心智。

推荐执行形状

概念代码：

class KnowledgeLibrary:
    def __init__(self, pool, embedding_generator):
        self._pool = pool
        self.embedding_generator = embedding_generator

    async def search(self, query: str, ...):
        embedding = await self.embedding_generator.generate(query)

        async with self._pool.connection() as conn:
            async with conn.cursor() as cur:
                await cur.execute(...)
                rows = await cur.fetchall()

        return ...

这里两个 async with 分别表达：

pool.connection()
→ acquire / release connection lease

conn.cursor()
→ acquire / close cursor

数据库等待通过 async driver 表达，Event Loop 才能调度其他 Task。

Async Pool 的目标不是让 SQL 更快

这句话很值得记：

AsyncConnectionPool 不会让 PostgreSQL 本身更快；它让等待 PostgreSQL 的时间不占住 Event Loop，并用受控数量的连接服务多个 concurrent requests。

如果 SQL 本身需要 2 秒：

async pool

不会把它变成 20ms。

它解决：

resource reuse
+ cooperative waiting
+ bounded concurrency

查询优化仍然是另一层问题。

FastAPI Lifespan 为什么是自然 Resource Owner

推荐：

startup
├─ initialize/open checkpoint pool
├─ initialize/open KnowledgeLibrary pool
├─ bounded connectivity check
└─ mark ready

requests
└─ checkout/release leases

shutdown
├─ stop new work
├─ close KnowledgeLibrary pool
└─ close checkpoint pool

比 module-level singleton 好在：

• startup failure 早暴露；
• readiness 可测试；
• shutdown ownership 明确；
• 测试可以 inject fake/test pool；
• 不依赖 hidden lazy global state。

Startup 也需要 Timeout

异步不代表可以无限 await。

如果 DB DNS/auth/route 出问题：

open pool
→ wait forever

同样是生产故障。

应有：

open / wait ready within T
→ success
or
→ explicit startup failure

这样 orchestrator 才知道容器未 ready，而不是“启动成功但每个请求都超时”。

Pool Capacity 是系统级配置

假设：

4 Uvicorn worker processes
× each pool max 10
= up to 40 DB connections

因此 pool size 不能只在一个 Python class 里拍脑袋。

要同时考虑：

• PostgreSQL max_connections；
• 其他服务连接预算；
• process/worker 数；
• interactive search 并发；
• ingestion/background jobs；
• query duration；
• SLO/tail latency。

所以：

pool max
= capacity policy

不是“越大越快”。

Pool Exhaustion 是 Backpressure

如果 pool max=5，20 个 request 同时来：

5 个 acquire connection
15 个 await availability

后 15 个 coroutine 没有阻塞 Event Loop，只是在异步排队。

这是正常 backpressure，但 latency 会增加。

因此应观测：

pool in-use
pool wait duration
acquire timeout
database query latency
request deadline

[!important] non-blocking wait ≠ no waiting。

Search 与 Ingestion 的 Transaction 语义不同

Search

通常：

short
read-only
one logical query operation

适合短 checkout。

Ingestion

一个 GeneratedKnowledgePack 可能同时写：

knowledge_sources
knowledge_segments
knowledge_units
knowledge_clips

它们共同描述一个逻辑知识包。

理想语义通常是：

全部成功
or
全部 rollback

所以：

async with pool.connection() as conn:
    async with conn.transaction():
        await insert_source(...)
        await insert_segments(...)
        await insert_units(...)
        await insert_clips(...)

不要为了“每个 DAO 都自己 async”拆成多个独立 autocommit transaction。

Async Refactor 最大的隐藏回归：Event Loop 好了，Atomicity 坏了

旧同步代码可能至少保证：

BEGIN
source
segments
units
clips
COMMIT

如果重构后变成：

Connection A → source → commit
Connection B → segments → commit
Connection C → units → ERROR

系统虽然“不阻塞”了，却留下半个 knowledge pack。

这是典型：

execution correctness improved
business correctness regressed

所以 L4 async refactor 必须带 transaction regression tests。

Delete-then-Insert Overwrite 也要放在一个 Intentional Transaction 里审查

旧 ingestion 可能：

find existing source
→ delete old source
→ commit
→ later insert new pack

如果新 pack 插入失败，旧数据已经消失。

这类 overwrite semantics 应明确：

replace atomically?
versioned append?
soft delete?

Async migration 是一个好机会把原本模糊的 transaction boundary 显式化，而不是机械逐行加 await。

pgvector Registration 也必须跟 Connection Lifecycle 对齐

旧同步路径：

register_vector(conn)

Async pool 中每个新物理连接都可能需要 async-compatible registration：

await register_vector_async(conn)

关键问题是注册时机：

pool creates new physical connection
→ initialize session/type adapters
→ then lease to application

不能只对某一个临时 connection 注册后，就假设 pool 中所有 connection 都有 vector adapter。

Connection Configure Hook 的心智

一个 pool 往往允许：

on-new-connection configure

可以统一设置：

• pgvector type registration；
• statement/session options；
• tracing metadata；
• application name。

这样：

every physical connection enters pool in valid initialized state

比业务 method 每次记得初始化更可靠。

Async Context Manager 与 Transaction Context Manager 不要混

async with pool.connection() as conn:

主要控制 connection lease。

async with conn.transaction():

主要控制 commit/rollback transaction scope。

两者职责不同。

“出了 connection() context”不应该被当成你所有业务 transaction semantics 的唯一表达。

Query Cancellation / Timeout

Native async driver 的价值之一是等待可以参与 cancellation。

例如 request deadline 到达：

coroutine cancelled
→ DB await interrupted/cancelled according to driver semantics
→ lease returned/cleaned

但仍要测试：

• cancellation 后 connection 是否可复用；
• transaction 是否 rollback；
• timeout 是否污染 pool。

不能只测试 happy path。

Connection Failure 要有明确错误边界

常见 failure：

pool startup fail
acquire timeout
query timeout
connection reset
transaction serialization/deadlock

它们不应该都被压成：

KnowledgeLibrary returned []

因为：

No evidence found
≠
Retrieval infrastructure failed

这对 Agent DecisionAuthority 很重要：EvidenceGap unresolved 的原因应可区分。

Empty Search 与 Infrastructure Error 必须分开

如果 DB 正常查询：

0 rows

这是合法 retrieval observation。

如果 DB timeout：

query never completed

这不是“0 evidence”。

否则上层可能误判：

knowledge base has no answer

而真实根因是基础设施不可用。

所以 EvidenceAttempt 应能保存：

success-empty
vs
infrastructure-error

Injection 比 Hidden Singleton 更容易测试

更清晰的 composition：

FastAPI lifespan creates pool
↓
creates KnowledgeLibrary(pool, embedding_generator)
↓
routes/tools receive that capability

测试可以注入：

• fake pool；
• local test PostgreSQL pool；
• fault-injection connection；
• fake embedding generator。

而不是每个 test 都去 reset module global singleton。

Current Code Review 的关键结论

BodySense 当前/审查基线里的 KnowledgeLibrary 具有：

async public methods
+
sync psycopg connection/cursor

这就是典型：

async façade around blocking I/O

修复的 north-star 不是“多包一层 async function”，而是把 resource boundary 真正迁移到 async driver/pool。

测试矩阵

Pool Lifecycle

app start → pool open once
app stop  → pool close once

Concurrent Search

并发 search 可以独立 checkout lease，不再触碰一个 sync singleton connection。

Heartbeat

slow DB fake / integration query 期间，async heartbeat 仍可运行。

Transaction Rollback

在插入 units/clips 中途制造异常：

source/segments/units/clips
→ no partial logical pack committed

pgvector Initialization

新物理 connection 能正确接受 vector 参数/结果。

Startup Failure

DB 不可达时在 bounded timeout 内显式失败。

Acquire Timeout

pool exhausted 时 request 进入可识别 timeout/backpressure path。

Semantic Parity

Async 重构前后仍保持：

• filters；
• vector distance semantics；
• candidate limit；
• intent boost；
• clip loading；
• top_k ordering contract。

Error Semantics

0 rows
≠ DB failure

上层 EvidenceAttempt 能区分。

常见误区

“用了 Pool 就 Async 了”

同步 pool 仍然会阻塞 Event Loop。关键是 async driver + awaitable query/acquire。

“Event Loop 单线程，所以一个 DB Connection 足够”

错。多个 Task 可以并发等待 DB，而且 transaction/session 语义也不允许把所有业务操作随意塞进同一个长期 connection。

“每次新建连接最干净”

语义简单，但在热路径上重复付 handshake/connection cost。

“async transaction 会自动跨多个 connection”

不会。一个 atomic transaction 必须明确绑定一个 transaction/connection scope。

“pool wait 是 blocking”

Async acquire 的等待是 cooperative backpressure，不会占住 loop thread；但仍会增加 latency。

“DB error 返回 [] 更鲁棒”

这会把 infrastructure failure 伪装成 no evidence，破坏上层 EvidenceGap / Failure Attribution。

自测题

1. 为什么长期 singleton connection 不是合理的 async concurrency model？
2. 为什么“每次请求一个 AsyncConnection”也有明显问题？
3. Pool、Connection Lease、Transaction 三种 lifetime 如何区分？
4. Async pool 为什么不会让 SQL 自身更快？
5. Pool exhaustion 为什么叫 backpressure，不叫 Event Loop blocking？
6. 多表 ingestion 为什么不能在 async refactor 中拆成多个独立 commit？
7. pgvector registration 为什么要绑定每个新 physical connection 的 lifecycle？
8. 为什么 0 rows 和 DB timeout 不能映射成同一个结果？
9. Startup readiness 为什么比 first-request lazy connect 更可靠？
10. 如何用 transaction rollback + heartbeat 两类测试同时证明业务正确性和 async 正确性？

最终不变量

Pool lifetime = application lifespan
Connection lifetime = short lease
Transaction lifetime = domain atomic operation
DB wait = native async I/O
Capacity = bounded resource
Errors = distinguish no-result from infrastructure failure

只要这六条清楚，KnowledgeLibrary 才真正从“async 方法包 sync psycopg”变成 production-shaped 异步 RAG 数据访问层。