gRPC客户端到服务端单向流中服务端能否取消流并返回错误

核心结论 #

客户端到服务端的单向流（客户端流式RPC）场景下，流接收方（服务端）完全可以主动取消流并向发送方返回错误、trailer元数据，不需要改用双向流，也不需要额外新增查询状态的RPC端点。当前代码无法获取返回内容，是对gRPC流的处理时序和API用法有误导致的。

问题原因 #

现有实现存在两个核心错误：

1. 服务端API使用错误
   Stream.SendAndClose()方法仅用于服务端完整接收完客户端所有流消息（即服务端Recv()返回io.EOF，确认客户端已经主动结束发送）后的正常响应场景。如果服务端要在接收流的过程中提前终止流，直接调用SendAndClose()不会触发框架立刻向客户端发送响应和trailer——gRPC框架只有在服务端流处理函数真正返回之后，才会把响应、状态、trailer刷写到传输层发给客户端。
2. 客户端读取时序错误
   Stream.Trailer()方法只有在流完全终止（即CloseAndRecv()返回结果，无论返回的是正常响应还是错误）之后，才能拿到完整的trailer元数据，在这之前调用只会返回空map。

正确实现方式 #

服务端逻辑 #

• 提前终止流时，不要调用SendAndClose()，直接从流处理函数返回一个通过status.Error()构造的gRPC状态错误即可，之前通过SetTrailer()设置的元数据会被框架自动携带发送给客户端。
• 只有在读完客户端所有消息（Recv()返回io.EOF）的正常场景下，才调用SendAndClose()返回成功响应。
  示例代码：

func (s *FooServer) Eat(stream pb.Foo_EatServer) error {
    // 提前设置要返回的trailer
    trailerMD := metadata.New(map[string]string{
        "error": "something went wrong",
    })
    stream.SetTrailer(trailerMD)

    for {
        food, err := stream.Recv()
        if err == io.EOF {
            // 客户端已发送完所有消息，走正常返回逻辑
            return stream.SendAndClose(&pb.Status{Status: "success"})
        }
        if err != nil {
            // 接收异常直接返回
            return err
        }
        // 业务校验，不符合要求则提前终止流
        if !validateFood(food) {
            // 直接返回带状态码的错误，框架会主动终止流并通知客户端
            return status.Error(codes.InvalidArgument, "invalid food param")
        }
    }
}

客户端逻辑 #

• Send()返回错误（包括io.EOF）时，立刻停止发送消息，调用CloseAndRecv()等待流最终结束。
• 必须等CloseAndRecv()返回之后，再调用Trailer()读取元数据，这时候无论请求成功还是失败，都能拿到服务端设置的trailer和状态信息。
  示例代码：

stream, err := fooClient.Eat(context.Background())
if err != nil {
    // 初始化流失败处理
    log.Fatalf("init stream failed: %v", err)
}

for _, stuff := range sendList {
    err = stream.Send(stuff)
    if err != nil {
        // 发送失败，说明流已被服务端终止或断连，停止发送
        break
    }
}

// 无论Send是否报错，都必须调用CloseAndRecv获取最终状态
resp, err := stream.CloseAndRecv()
// CloseAndRecv返回后再读取trailer，此时可以拿到完整元数据
md := stream.Trailer()

if err != nil {
    // 从错误中解析服务端返回的gRPC状态
    if st, ok := status.FromError(err); ok {
        log.Printf("rpc failed, code: %v, msg: %s", st.Code(), st.Message())
    }
    log.Printf("trailer from server: %v", md)
    return
}
// 正常处理响应
log.Printf("rpc success, resp: %v, trailer: %v", resp, md)

补充说明 #

服务端从处理函数返回错误后，gRPC底层会自动通过HTTP2的RST_STREAM帧通知客户端流已终止，客户端后续的Send()调用会立刻返回错误，不需要额外做流取消的处理，整个流程符合gRPC标准规范，没有兼容性问题。

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