NestJS代理流在Cloud Run无法流式传输（缓冲至结束）

问题背景 #

本地环境中，NestJS代理Cloud Run上的FastAPI流式响应时，客户端能实时分块接收数据；但部署到Cloud Run后，响应会被完全缓冲，直到上游流结束才一次性交付。已尝试设置流式请求头、禁用压缩、添加心跳、切换HTTP/1.1等方案，均无效。

核心原因 #

Cloud Run的前端Envoy代理默认会缓冲响应内容，直到达到32KB阈值或上游连接关闭才会转发给客户端；同时NestJS底层Express的响应对象存在内部缓冲，会进一步延迟数据发送。

具体解决步骤 #

1. 强制禁用Express的响应缓冲 #

Express的ETag机制需要计算完整响应内容，会触发全局缓冲，必须显式禁用；同时绕过Express的res.write，直接操作底层TCP Socket发送数据，避免中间层缓冲：

// 在设置响应头时添加
res.setHeader('ETag', ''); // 禁用ETag，阻止缓冲触发
// 替换原res.write逻辑，直接写入socket
if (res.socket?.writable) {
  res.socket.write(JSON.stringify(obj) + '\n');
}

2. 改用原生HTTP模块请求上游 #

Axios的responseType: 'stream'存在内部缓冲逻辑，改用Node.js原生http/https模块请求上游，能更精准控制流的实时传输：

const url = new URL(`${process.env.UPSTREAM_URL}/stream_query`);
const httpModule = url.protocol === 'https:' ? require('https') : require('http');

const upstreamReq = httpModule.request({
  method: 'POST',
  hostname: url.hostname,
  port: url.port,
  path: url.pathname + url.search,
  headers: {
    'Content-Type': 'application/json',
    Authorization: `Bearer ${token}`,
    Accept: 'application/json',
  },
}, (upstreamRes) => {
  upstreamRes.pipe(transform);
});

upstreamReq.write(JSON.stringify(/* 请求体内容 */));
upstreamReq.end();

// 处理上游请求错误
upstreamReq.on('error', (err) => {
  console.error('上游请求失败:', err);
  clearInterval(heartbeat);
  res.status(500).end();
});

3. 优化心跳机制 #

原心跳用SSE格式，改为NDJSON统一格式，确保心跳块足够小（远小于32KB），强制Cloud Run实时转发：

const heartbeat = setInterval(() => {
  if (res.socket?.writable) {
    res.socket.write(JSON.stringify({ type: 'heartbeat' }) + '\n');
  }
}, 2000);

4. 确认Cloud Run配置 #

• 部署时全局禁用响应压缩，避免代理层自动缓冲压缩内容
• 设置足够长的服务超时时间，防止流传输过程中被中断
• 无需调整并发数，但要确保服务实例资源充足，不会因负载过高导致缓冲

调整后的完整代码示例 #

async streamQuery(res: Response) {
  // 配置流式响应头，禁用缓冲相关机制
  res.removeHeader('Content-Length');
  res.setHeader('Transfer-Encoding', 'chunked');
  res.setHeader('Content-Type', 'application/x-ndjson; charset=utf-8');
  res.setHeader('Cache-Control', 'no-store, no-transform');
  res.setHeader('X-Accel-Buffering', 'no');
  res.setHeader('Connection', 'keep-alive');
  res.setHeader('Content-Encoding', 'identity');
  res.setHeader('ETag', ''); // 关键：禁用ETag
  res.flushHeaders();

  // 心跳机制（NDJSON格式）
  const heartbeat = setInterval(() => {
    if (res.socket?.writable) {
      res.socket.write(JSON.stringify({ type: 'heartbeat' }) + '\n');
    }
  }, 2000);

  const transform = new Transform({
    transform(chunk, _encoding, callback) {
      try {
        const line = chunk.toString().trim();
        if (line) {
          const obj = JSON.parse(line.startsWith('data: ') ? line.slice(6) : line);
          // 直接写入socket，绕过Express缓冲
          if (res.socket?.writable) {
            res.socket.write(JSON.stringify(obj) + '\n');
          }
        }
      } catch (err) {
        console.error('解析失败:', err);
      }
      callback();
    },
    flush(callback) {
      clearInterval(heartbeat);
      if (res.socket?.writable) {
        res.socket.end();
      }
      callback();
    },
  });

  // 原生HTTP请求上游
  const url = new URL(`${process.env.UPSTREAM_URL}/stream_query`);
  const httpModule = url.protocol === 'https:' ? require('https') : require('http');
  
  const upstreamReq = httpModule.request({
    method: 'POST',
    hostname: url.hostname,
    port: url.port,
    path: url.pathname + url.search,
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${token}`,
      Accept: 'application/json',
    },
  }, (upstreamRes) => {
    upstreamRes.pipe(transform);
  });

  upstreamReq.write(JSON.stringify(/* 请求体内容 */));
  upstreamReq.end();

  upstreamReq.on('error', (err) => {
    console.error('上游请求出错:', err);
    clearInterval(heartbeat);
    res.status(500).end();
  });
}

额外注意事项 #

• 使用Node.js 16+版本，避免旧版流处理的兼容性问题
• 如果用Fastify作为NestJS适配器，需禁用Fastify的全局压缩，并通过reply.raw操作原生Socket
• 用curl -v测试响应，确认头信息包含Transfer-Encoding: chunked，且数据分块到达

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