MagicStack / MagicStack/asyncpg

Connection.copy_records_to_table hanging when generator passed to records parameter raises exception with long error message

オープン
#1,354 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

まだ誰も着手していません。

主要言語
Python
スター
8.1k
フォーク
468
PR マージ指標
30日以内にマージされた PR はありません

説明

When a generator passed to the records argument of Connection.copy_records_to_table raises an exception that when stringified with str(exc) produces a 9996 byte long (or greater) message it makes the COPY hang forever.

Reproduction below

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = "==3.15"
# dependencies = ["asyncpg==0.31.0"]
# ///
"""Repro: asyncpg deadlocks in ROLLBACK when the records generator passed to
copy_records_to_table raises an exception whose str() exceeds PostgreSQL's
10,000-byte frontend message limit.

Run (assumes an existing PostgreSQL; only touches a session-local temp table):

    DSN=postgres://user:pass@host:5432/db uv run repro.py

Exit code 1 means the deadlock occurred.
"""

import asyncio
import os
import sys

import asyncpg

DSN = os.environ.get("DSN", "postgres://postgres:postgres@localhost:5432/postgres")
HANG_TIMEOUT = float(os.environ.get("HANG_TIMEOUT", "15"))

# Length in bytes of the error message raised by the record generator.
# > 9995 bytes => PostgreSQL rejects the CopyFail frame ("invalid message
# length") and the deadlock triggers; at or below, the run fails cleanly.
# The threshold is exact: the frame's int32 length field (4 bytes, counted in
# itself) + payload + NUL terminator must not exceed PQ_SMALL_MESSAGE_LIMIT
# (10,000), so 10000 - 4 - 1 = 9995.
ERROR_BYTES = int(os.environ.get("ERROR_BYTES", "9996"))


def records():
    for i in range(100):
        yield (i,)
    raise ValueError("A" * ERROR_BYTES)


async def run_case():
    conn = await asyncpg.connect(DSN)
    try:
        async with conn.transaction():
            await conn.execute("CREATE TEMP TABLE repro_t (b int)")
            await conn.copy_records_to_table("repro_t", records=records())
    except ValueError:
        # clean failure, bug not reproduced
        pass
    finally:
        await conn.close()


async def main():
    task = asyncio.create_task(run_case())
    _, pending = await asyncio.wait([task], timeout=HANG_TIMEOUT)
    if pending:
        print("Statement timed out, bug reproduced", file=sys.stderr)
        return 1
    print("Statement executed before timeout, bug not reproduced", file=sys.stderr)
    return 0


if __name__ == "__main__":
    os._exit(asyncio.run(main()))

コントリビューションガイド

このリポジトリのコントリビューションガイドは索引されていません

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

調査の方向性

Connection.copy_records_to_tableから開始し、PostgreSQLに対してrepro.pyを実行して、9995バイトのしきい値付近でERROR_BYTESを変化させます。ジェネレーター例外とCOPYロールバックの経路を追跡します。しきい値以上のメッセージを含め、例外がハングせずに正常に完了すれば完了です。

索引モデルが issue の本文から書いたものです。

評価

技術スタック
postgresql, python
領域
databases
issue の種類
バグ
難易度
4/5
見積もり時間
3〜5日
活発さ
活発
明瞭さ
おおむね明確
初心者へのやさしさ
48/100

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。