MagicStack / MagicStack/asyncpg

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

Open
#1,354 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
Python
Stars
8.1k
Forks
468
PR merge metrics
No merged PRs in 30d

Description

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

```python
#!/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()))
```

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start at Connection.copy_records_to_table and run the supplied repro.py against PostgreSQL, varying ERROR_BYTES around the 9995-byte threshold. Trace the generator-exception and COPY rollback path; done means the exception completes cleanly without hanging, including for messages at or above the threshold.

Written by the indexing model from the issue text.

Assessment

Tech stack
postgresql, python
Domain
databases
Issue type
Bug
Difficulty
4/5
Estimated time
3-5 days
Activity status
Active
Clarity
Mostly clear
Newbie friendliness
48/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.