PostgreSQL idle-in-transaction timeout错误排查及测试咨询

问题背景 #

现有两个搜索函数实现，功能可正常运行，但频繁抛出以下错误：

OperationalError: (psycopg2.OperationalError) terminating connection due to idle-in-transaction timeout

需排查错误原因，并了解如何在开发环境测试连接闲置场景。

现有搜索函数实现 #

def search(cls: Type[T], session, search_query):
        combined_search_vector = (
            x.y.search_vector
            | sa.func.coalesce(xClass.yClass.search_vector, "")
        )

        with session as ctx_session:
            query = (
                ctx_session.query(Class)
                .join(xClass.YClass)
                .filter(combined_search_vector.match(f"'{search_query}'"))
            )

        return query

最初的搜索函数实现 #

@classmethod
    def search(cls: Type[T], session, search_query):
        combined_search_vector = (
            x.y.search_vector
            | sa.func.coalesce(xClass.yClass.search_vector, "")
        )

        try:
            query = (
                ctx_session.query(Class)
                .join(xClass.YClass)
                .filter(combined_search_vector.match(f"'{search_query}'"))
            )

        except sa_exc.DatabaseError:
            session.rollback()

        return query

会话配置代码 #

sessionmaker = attr.ib(default=None, repr=False)
    _session = attr.ib(default=None, repr=False)
    _engine = attr.ib(repr=False)
    _search_index_prefix = "search_"

    @_engine.default
    def create_engine(self):
        url = self.config["core"]["url"]
        engine = create_engine(url, echo=False)
        self._sessionmaker = sessionmaker(bind=engine)
        return engine

    @property
    def session(self):
        if self._session is None:
            self._session = self._sessionmaker()
            self._session.execute("SET pg_trgm.similarity_threshold = 0.2;")
        return self._session

    def close(self):
        if self._session is not None:
            self._session.close()
            self._session = None

错误原因排查 #

核心问题：事务生命周期管理缺失 #

PostgreSQL的idle-in-transaction timeout是指事务开启后长期无操作，数据库主动终止连接释放资源。你的问题根源在应用层会话/事务管理，而非搜索函数写法：

1. 单例会话长期持有：session属性采用单例模式，第一次创建后持续复用。一旦会话执行了execute（设置参数）或查询操作，会隐式开启事务，若未主动提交/回滚，事务会一直处于开启状态。
2. 事务未正确终结：
   • 最初版本仅在捕获异常时回滚，正常执行后事务未关闭，会话长期处于闲置事务状态。
   • 现有版本用with session as ctx_session仅创建了查询对象，实际查询在外部执行，事务仍处于开启状态，后续未及时终结就会触发超时。
3. 对会话管理的误解：PostgreSQL仅负责终止闲置事务连接，事务的开启、提交/回滚责任在应用层，而非数据库自动处理。

搜索函数写法的影响 #

两个版本的搜索函数本身不会直接触发超时，但都未处理事务终结：

• 最初版本完全缺失正常流程的事务关闭逻辑，问题更突出。
• 现有版本的上下文管理器未覆盖查询执行环节，事务仍会挂起。

修复方案 #

1. 显式管理事务生命周期 #

将查询执行过程包裹在事务上下文内，确保执行后自动终结事务：

def search(cls: Type[T], session, search_query):
    combined_search_vector = (
        x.y.search_vector
        | sa.func.coalesce(xClass.yClass.search_vector, "")
    )

    with session.begin():  # 显式开启事务，退出时自动提交/回滚
        query = (
            session.query(Class)
            .join(xClass.YClass)
            .filter(combined_search_vector.match(f"'{search_query}'"))
        )
        results = query.all()  # 在此执行查询，而非返回未执行的query对象

    return results

2. 调整会话管理模式 #

避免单例会话长期持有，改为按需创建和销毁：

@property
def session(self):
    # 每次调用返回新会话，由外部使用上下文管理器管理生命周期
    session = self._sessionmaker()
    session.execute("SET pg_trgm.similarity_threshold = 0.2;")
    return session

外部使用时：

with db.session as session:
    results = Class.search(session, "your_query")

3. 数据库配置调整（可选） #

若业务需长事务，可修改PostgreSQL的idle_in_transaction_session_timeout参数延长超时时间，但这是治标方案，优先修复应用层事务管理。

开发环境测试连接闲置场景 #

1. 快速触发超时：
   • 修改PostgreSQL配置文件postgresql.conf，设置idle_in_transaction_session_timeout = 1000（1秒），重启数据库。
   • 在代码中开启会话执行操作后，休眠超过1秒再复用会话执行新操作，即可触发超时错误。
2. 代码模拟：

from time import sleep
from your_module import db

session = db.session
# 执行操作开启事务
session.execute("SELECT 1")
# 休眠超过超时时间
sleep(2)
# 再次操作触发错误
try:
    session.query(Class).first()
except Exception as e:
    print(e)  # 会输出idle-in-transaction timeout错误

3. 监控闲置事务：
   • 打开psql执行以下语句，查看当前处于闲置事务状态的连接，确认应用会话是否在列：

   SELECT pid, state, query FROM pg_stat_activity WHERE state = 'idle in transaction';
   

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