COPY from stdin failed: PgCopyIn dropped without calling finish() or fail()

主线位置：查询主线旁边的 COPY 支路，坏在 PgConnection::run 开头的 wait_until_ready 上，经连接池时还牵扯到连接池主线归还时的 ping。COPY 怎么收发消息、Drop 为什么只能排一条 CopyFail，见LISTEN/NOTIFY 与 COPY；本页只讲现象、范围和修法。

flowchart TB
  drop["drop PgCopyIn"] --> queued["CopyFail 进写缓冲"]
  queued --> q1["下一个查询"]
  q1 --> err["报 57014，只报这一次"]
  err --> off["线上多出一组回应"]
  off --> shift["之后每个查询读到上一个的结果"]
  queued -.经 Pool 借出.-> ping["归还时 ping 报错"]
  ping --> closed["连接被关掉，没事"]

本页实验在 M2 Max、macOS、rustc 1.99.0、Postgres 16.14 上运行，依赖 sqlx = "=0.9.0"；同一个程序换成 sqlx = "=0.8.6" 输出逐字相同。

现象

一个导入函数在 copy_in_raw 和 finish 之间用 ? 提前返回了（输入校验失败、上游读出错），PgCopyIn 就在既没有 finish 也没有 abort 的情况下被 drop。之后这条连接上的第一个查询报错：

error returned from database: COPY from stdin failed: PgCopyIn dropped without calling finish() or fail()

错误码是 57014（query_canceled）。看起来是「报了错，提醒你一下」，问题在后面：从第二个查询起，结果全部错开一位，而且是 Ok。

最小复现

src/bin/copy_drop.rs 的 A 段，一条直接持有的 PgConnection：

let mut conn = PgConnection::connect_with(&opts()?).await?;
let mut copy = conn.copy_in_raw("copy item from stdin (format csv)").await?;
copy.send(b"1,apple\n".as_slice()).await?;
drop(copy);
for i in 1..=4 {
  // select $1 * 100，打印 Ok(v) 或错误
  probe("A", i, &mut conn).await;
}

--- A: PgConnection 上 drop ---
A select 1 * 100 -> Database(Some("57014"), COPY from stdin failed: PgCopyIn dropped without calling finish() or fail())
A select 2 * 100 -> RowNotFound
A select 3 * 100 -> Ok(200)
A select 4 * 100 -> Ok(300)

第 3 次查的是 3 * 100，拿到的是 200。连接不会自己恢复。

连接从池里借出来、在 PoolConnection 上做 COPY，然后接着用它，再还回池里（C 段，max_connections(1) 的池，这条连接上 select $1 * 100 已经在语句缓存里）：

--- C: pool.acquire() 拿到的连接上 drop，接着用这条连接，再还回去 ---
C select 1 * 100 -> Database(Some("57014"), COPY from stdin failed: PgCopyIn dropped without calling finish() or fail())
C select 2 * 100 -> Ok(100)
C select 3 * 100 -> Ok(200)
C 还回去之后 pool.size() = 1
C 从池里 select 4 * 100 -> RowNotFound
C 从池里 select 5 * 100 -> RowNotFound
C 从池里 select 6 * 100 -> RowNotFound

在借出的连接上，还是错开一位。还回池里之后更糟：每次查询都是 RowNotFound。换成 execute 就是「影响 0 行」，同样不报错。

原因

PgConnection 用 pending_ready_for_query_count 记还欠着几条 ReadyForQuery。每发一个 Sync，write_sync 就加一；wait_until_ready 先 flush 写缓冲，再读消息、读到一条 ReadyForQuery 减一，减到 0 就认为线上已经干净了。

PgCopyIn 的 Drop 不能 await，只把一条 CopyFail 写进缓冲。服务端对它会回 ErrorResponse 加 ReadyForQuery，可计数没有加一（Drop 没走 queue_simple_query）。于是：

1. 第 1 个查询：wait_until_ready 看到计数是 0，flush 完 CopyFail 就返回；这个查询的第一条回应是 CopyFail 引出的 ErrorResponse，报 57014。它自己的那组回应还在线上。
2. 第 2 个查询：wait_until_ready 读到 COPY 的那条 ReadyForQuery，计数归零，停下；接下来读到的是第 1 个查询的回应。A 段里第 1 个查询发的是 Parse（语句还没缓存），它的回应里没有 DataRow，所以是 RowNotFound；C 段里语句已缓存，第 1 个查询发的是 Bind + Execute，回应里有 DataRow，所以拿到 100。
3. 此后每次都多出一组、少读一组，偏移永远是一。

逐条协议消息的走法在LISTEN/NOTIFY 与 COPY的动脑筋答案里。

还回池里之后永远 RowNotFound，是因为池在借出和归还时各发一次 ping（ping 就是 write_sync + wait_until_ready，见获取连接和归还、Drop 与维护任务）。错开一位以后，借出时的 ping 读掉的是上一次查询的回应，查询读到的是这次 ping 的那条光秃秃的 ReadyForQuery，归还时的 ping 又把查询自己的回应读掉。经代理看（TRACE=1）：

  [conn 2] C->S Sync
  [conn 2] C->S Bind(portal unnamed, stmt sqlx_s_1; [8] 00 00 00 00 00 00 00 04), Execute, Close(P, unnamed), Sync
  [conn 2] S->C ReadyForQuery(I)
C 从池里 select 4 * 100 -> RowNotFound
  [conn 2] C->S Sync
  [conn 2] S->C BindComplete, DataRow(列长 8), CommandComplete("SELECT 1"), CloseComplete, ReadyForQuery(I)
  [conn 2] S->C ReadyForQuery(I)

每个来回三次 Sync、读三条 ReadyForQuery，计数始终对得上，所以连接永远不会因为计数出错而被丢掉。

--- B: Pool::copy_in_raw 后 drop ---
 WARN error occurred while testing the connection on-release error=error returned from database: COPY from stdin failed: PgCopyIn dropped without calling finish() or fail() at line 315
B pool.size() = 0
B select 1 * 100 -> Ok(100)
B select 2 * 100 -> Ok(200)

修法

每条路径都调到 finish 或 abort。 不要让 ? 把 PgCopyIn 带出作用域；自己的校验失败时先 abort，再返回错误：

async fn load(conn: &mut PgConnection, lines: &[&str]) -> anyhow::Result<u64> {
  let mut copy = conn.copy_in_raw("copy item from stdin (format csv)").await?;
  for line in lines {
    if line.is_empty() {
      copy.abort("empty line").await?;
      anyhow::bail!("输入里有空行，已 abort");
    }
    copy.send(format!("{line}\n").into_bytes()).await?;
  }
  Ok(copy.finish().await?)
}

send 自己返回错误时 copy 也会随 ? 被 drop，这条路径归下一条修法管。

已经 drop 了，就别再用这条连接。 直接持有的 PgConnection 就 close()；PoolConnection 调 close_on_drop()，或者 detach() 之后自己关。D 段在 drop 了 PgCopyIn 之后调 close_on_drop：

let mut pc = pool.acquire().await?;
let mut copy = pc.copy_in_raw("copy item from stdin (format csv)").await?;
copy.send(b"3,cherry\n".as_slice()).await?;
drop(copy);
pc.close_on_drop();
drop(pc);

--- 修法：出错的连接 close_on_drop ---
D pool.size() = 0
D 从池里 select 1 * 100 -> Ok(100)
D 从池里 select 2 * 100 -> Ok(200)
--- 修法：错误路径上 abort ---
load -> Err("输入里有空行，已 abort")
fix select 1 * 100 -> Ok(100)
fix select 2 * 100 -> Ok(200)
load -> Ok(2)
fix select 3 * 100 -> Ok(300)

不需要在原连接上接着干活时，用 pool.copy_in_raw。 上面「没有蠢问题」里说了，这种写法下 drop 的后果是关掉一条连接，而不是错位。

怎么确认自己没中招

• 搜代码里的 copy_in_raw：看从它到 finish / abort 之间有没有 ?、return、break，或者会被取消的 .await（tokio::select!、超时）。PgCopyIn 标了 #[must_use]，但只管「完全没用返回值」，管不到提前返回。
• 日志里出现过一次 PgCopyIn dropped without calling finish() or fail()，就说明那条连接之后的结果都不可信；只出现在 error occurred while testing the connection on-release 这条 WARN 里的，连接已经被关掉了。
• 上游没有针对「drop 之后错位」的 issue（搜过 PgCopyIn drop、copy_in_raw drop、PgCopyIn dropped without calling finish、CopyFail）。相近的有：#4469 指出 COPY 发 Query 时不给计数加一，并提到 finish() 被取消会让池里的连接「one reply behind」；#4454 修 COPY TO STDOUT 的同类问题。到本文写作时两者都没有合并。

小结

• PgCopyIn 被 drop 时只排一条 CopyFail，它的 ReadyForQuery 不进 pending_ready_for_query_count：下一个查询报 57014，之后每个查询读到上一个查询的结果。
• 经连接池借出、错位后还回池里的连接，之后每次查询都是 RowNotFound（execute 是 0 行），因为借出和归还的 ping 各吞掉一组回应。
• pool.copy_in_raw 返回的 PgCopyIn 被 drop 时，归还的 ping 失败，连接被关掉，没有错位。
• 修法：每条路径都 finish 或 abort；已经 drop 了就关掉这条连接（close、close_on_drop、detach）。0.8.6 和 0.9.0 行为相同。