Go io.Pipe 的错误传播：生成失败与消费失败怎样让两端退出

生成压缩内容的一端已经失败，上传端却只读到正常 EOF；反过来，上传端不想再读了，生成 goroutine 还堵在 Write。这两种故障都不是加一层缓冲区就能解决的，真正缺的是两端对结束原因与退出责任的约定。

io.Pipe 把同步写入与读取配对起来。它没有一份可以无限容纳内容的内部队列，写端通常要等数据被读取以后才能继续。因此它很适合连接生成器与流式消费者，也很容易在另一端提前退出时留下等待者。

下面固定 Go 1.24.1，用标准库构造三条路径：正常完成、生成失败、消费失败。所有 goroutine 都有明确等待者，错误用 errors.Is 判断，不靠一段打印文字猜有没有传播成功。

两端各自拥有一个关闭动作

写端生成结束，负责关闭 PipeWriter。正常结束用无错误关闭；生成失败则把原因交给 CloseWithError，让读端继续读取时得到真正的错误。

读端不再消费时，负责关闭 PipeReader。若因为下游失败停止，应把下游错误交给读端的 CloseWithError，让仍在写入的生产者也能解除阻塞。不要等生产者自己碰巧写完：它可能正等着那个已经退出的消费者。

保存 main.go：

package main

import (
    "errors"
    "fmt"
    "io"
    "strings"
)

var errSource = errors.New("generator failed")
var errSink = errors.New("destination rejected write")

func produce(writer *io.PipeWriter, fail bool) (err error) {
    defer func() { _ = writer.CloseWithError(err) }()
    for index := 1; index <= 3; index++ {
        if fail && index == 2 {
            return errSource
        }
        if _, err = fmt.Fprintf(writer, "row %d\n", index); err != nil {
            return err
        }
    }
    return nil
}

type rejectingWriter struct{}

func (rejectingWriter) Write(data []byte) (int, error) {
    return 0, errSink
}

func transfer(failSource, failSink bool) (string, error, error) {
    reader, writer := io.Pipe()
    finished := make(chan error, 1)
    go func() { finished <- produce(writer, failSource) }()

    var result strings.Builder
    var destination io.Writer = &result
    if failSink {
        destination = rejectingWriter{}
    }
    _, copyErr := io.Copy(destination, reader)
    // 先解除可能还在等待读端的生产者，再等它退出。
    _ = reader.CloseWithError(copyErr)
    producerErr := <-finished
    return result.String(), copyErr, producerErr
}

func main() {
    text, readErr, producerErr := transfer(false, false)
    if text != "row 1\nrow 2\nrow 3\n" || readErr != nil || producerErr != nil {
        panic("normal transfer failed")
    }
    fmt.Println("normal: all rows, both sides exited")

    text, readErr, producerErr = transfer(true, false)
    if text != "row 1\n" || !errors.Is(readErr, errSource) ||
        !errors.Is(producerErr, errSource) {
        panic("source error did not reach consumer")
    }
    fmt.Println("source failure: partial output, consumer received source error")

    _, readErr, producerErr = transfer(false, true)
    if !errors.Is(readErr, errSink) || !errors.Is(producerErr, errSink) {
        panic("sink error did not unblock producer")
    }
    fmt.Println("sink failure: producer unblocked, both sides exited")
}

执行 go run main.go，三条断言路径都应完成并输出相应说明。第二条保留了已经读出的第一行，但整体返回错误；第三条证明读端关闭后，生产者不再永远等待写入。

这里等待生产者之前先关读端，顺序不能反过来。若先执行 <-finished，生产者可能还在 PipeWriter.Write 中等待消费，而读端又在等待它结束，闭环就卡住了。退出动作的先后关系，比是否用了 channel 更重要。

正常 Close 会把失败包装成成功结束

最常见的错误写法是 goroutine 里 defer writer.Close()，内部生成函数返回错误以后只记一条日志。读端看到的是正常 EOF，io.Copy 可能返回 nil；调用者于是把半份内容当作成功产物。

用具名返回值配合 CloseWithError(err)，可以让当前函数选择的错误进入写端关闭流程。但这不是在给 Pipe 添加全局异常收集器：生产者自己吞掉错误、返回 nil，读端仍只能得到正常结束。因此生产函数的返回合同同样要准确。

Pipe 也不会替应用把两端错误自动合并成一条“最真实的错误”。如果消费失败导致生产者写入失败，两端可能得到同一个下游原因；若双方几乎同时独立失败，则应在业务层决定主错误和附加诊断，而不是简单按最后打印的一行覆盖前面的信息。

示例直接保留了消费错误和生产者错误，方便测试这两个方向。实际 API 可以定义自己的结果结构，或在适当版本下组合错误，但不能把生成错误仅当调试日志，继续对外返回成功。

关闭动作还要保持幂等思维。CloseWithError不会用后来的错误覆盖已经记录的同端关闭错误，返回nil也不说明对端业务成功。它表达的是关闭动作已经执行，而不是获得了一份远端确认。若多处都可能触发退出，应选定谁报告业务结果，其他关闭动作只承担解除阻塞和释放资源的职责。

代码里的完成通道只传一项，并且有一个槽位。生产者能够登记最终结果，然后结束；消费者仍会显式接收它，验证没有把尚在运行的工作丢到后台。把完成通道删掉以后即使主函数暂时没有卡住，也不能证明生产goroutine已经退出。

缓冲与取消是两个独立设计

给写端套 bufio.Writer，可能让少量数据先留在缓冲区里，写入暂时不阻塞，却没有改变对端消失以后的退出责任。最终刷新仍可能失败；忘记检查 Flush 的结果，还会再丢一层错误。

同样，context 被取消，不会让一个任意 io.Reader 自动停止。若生成器或消费者需要响应取消，可以安排取消时关闭相关 Pipe 端，解除阻塞，再等待 goroutine 结束。这个桥接动作需要有人拥有，不能只给函数签名加一个 ctx 参数就宣称完成。

对于在生成过程中长时间执行 CPU 运算的任务，关闭 Pipe 只能影响它下一次读写，不会安全地从任意指令位置中断。计算循环仍需检查退出信号，或按业务要求采用能够独立管理生命周期的执行方式。

实际上传还要区分网络请求等待结束和服务端业务提交。客户端消费失败后关闭 Pipe，能说明本地生成器已经停止，不能证明远端没有收到此前字节，也不能证明此前字节没有被持久化。重要写入需要稳定业务编号和结果查询，而不是把一次连接错误当成安全重试证据。

已经交出去的字节不会自动撤回

第二条路径故意保留第一行，提醒调用者：流失败以前，消费者已经拿到部分数据。若目标是正式文件，应先写暂存文件，完整复制成功后再发布；若目标是远程对象存储，应使用其明确的上传完成或终止协议。

io.Pipe 不是事务边界。它提供同步读写与错误通道，无法撤销消费方已经做的事情。要不要保留半份结果，应该由消费方的发布协议决定，不能悄悄藏在生成器里。

排查这类阻塞时，至少确认三个位置：谁还在等待 Write，谁决定不再 Read，谁负责把两端最终等完。只要每一端都知道自己何时关闭、用什么原因关闭，Pipe 才是连接两个有界生命周期的工具，而不是把一个错误藏进另一个 goroutine 的办法。