Spring ProtobufDecoder 丢截断尾帧：别把 Flux 完成当完整

接收带长度前缀的 Protobuf 消息流，Flux 已经正常完成，实际消息数却少了一条。若输入恰好在最后一帧中断，或者结束在尚未完成的 varint 长度前缀里，应检查解码器是否真正验证了流的完整性。

Spring 7.0.10 的 ProtobufDecoder 流式路径在结束时清理未完成缓冲，没有把“还缺字节”转成结束错误。另一个边界是单字节 0x00：它可以表示长度为零的默认消息，不等于完全没有输入。这些是受控输入下的解码边界，不能据此推断线上数据已经丢失或存在外部利用。

先确认传输约定确实有长度前缀

连续 Protobuf 消息需要边界信息。这里讨论的是每条消息前都有 varint 字节长度的格式，不是把多个裸消息直接拼接，也不是 gRPC 自己的五字节帧头。协议选错时，再完整的读取器也会把内容解释错。

核对完整单帧、最后一帧少字节、长度前缀只传一部分以及零长度帧。不要只用 HTTP 成功或 Flux 的完成信号，代替消息边界检查。

有限批次可以先验证完整帧，再交业务处理

对于总大小可控、允许全部完成后才交付的批次，可先有界读取原始字节，再逐帧读取声明的确切长度。下面是批次字节已经取得后的 Java 工具方法，使用标准 Protobuf 运行库；不是替换全部 WebFlux codec 的配置：

import com.google.protobuf.CodedInputStream;
import com.google.protobuf.ExtensionRegistryLite;
import com.google.protobuf.MessageLite;
import com.google.protobuf.Parser;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

static <T extends MessageLite> List<T> decodeFrames(
        byte[] bytes, Parser<T> parser, ExtensionRegistryLite registry,
        int maxFrameBytes, int maxMessages) throws IOException {
    if (maxFrameBytes <= 0 || maxMessages <= 0) {
        throw new IllegalArgumentException("positive limits required");
    }
    CodedInputStream input = CodedInputStream.newInstance(bytes);
    List<T> result = new ArrayList<>();
    while (!input.isAtEnd()) {
        if (result.size() >= maxMessages) {
            throw new IOException("too many messages");
        }
        int size = readFrameSize(input, maxFrameBytes);
        byte[] body = input.readRawBytes(size);
        result.add(parser.parseFrom(body, registry));
    }
    return result;
}

static int readFrameSize(CodedInputStream input, int maximum) throws IOException {
    long size = 0;
    for (int shift = 0; shift < 35; shift += 7) {
        int next = input.readRawByte() & 0xff;
        size |= (long) (next & 0x7f) << shift;
        if ((next & 0x80) == 0) {
            if (size > maximum) {
                throw new IOException("message size exceeds limit");
            }
            return (int) size;
        }
    }
    throw new IOException("message size prefix is too long");
}

放在类中使用，parser 与 registry 沿用目标消息的真实定义。读取前还必须设置整个响应的字节上限；这个函数的单帧上限不能抵消已经创建巨大 byte 数组的成本。WebFlux 可在专门入口使用有明确 maxByteCount 的 DataBuffer 聚合，并按其所有权规则释放缓冲；不要关闭全局 codec 容量限制。已有WebClient 聚合上限应继续保留。

长度前缀最多读取五个字节，用 long 累积后再对照正整数上限，拒绝超限或仍未终止的前缀，不能先缩成 int 再检查。它允许 80 00 这类冗余零编码；若协议要求最短编码，需另加规范性校验。readRawBytes 要求取足声明字节，尾帧不足会失败。长度为零时仍调用 parser；允许空消息的模型得到默认实例，有必填约束的模型仍应失败。不能为了保住零长度消息而跳过模型初始化校验。

无界流需要状态化结束检查

长期连接或大批次不适合整份聚合。此时应采用有界、可取消的增量解码实现，在完成时检查长度前缀和消息体都没有残留，并为错误与取消释放缓冲。只增加 collectList 不会修复已经被解码器丢掉的尾部。

完整验证之前不要把整批标成成功；若业务边读边产生副作用，还需明确中断后的幂等与恢复记录。保留旧解码配置和输入原件，先在专门入口切换，撤回时恢复入口选择；已交付的短批次仍要按消息身份核对，恢复代码不会补回遗漏。