fix(relay): never forward a tunneled body that lost frames (#2822)

When the relay drops mid-request, the prompt_async body frames can be lost. The tunnel host forwarded the request to loopback as an empty/truncated chunked body, which the server rejects with a bare 400 (empty response body) — the mobile app's 'Failed to send message (400)'. Host now buffers request bodies (<512KB) and forwards the complete body only once StreamEnd arrives; larger bodies still stream live. A new hasBody flag on the request head lets the host detect a body that delivered zero frames and abort it as an ambiguous transport failure (which the client already retries) instead of forwarding an empty body. A 15s body-delivery deadline converts stalled tunnels into clean aborts.
2026-08-12 16:54:23 +03:00
parent 9e43b9ae46
commit 6a2f0b8135
5 changed files with 274 additions and 31 deletions
@@ -45,6 +45,8 @@ Everything a client normally sends to the single OpenChamber origin:

 The host dispatcher restricts tunneled traffic to explicit path allowlists (one for HTTP, one for WS).

+Request bodies crossing the tunnel are buffered on the host and forwarded to loopback only once the client's `StreamEnd` frame arrives (bodies above ~512 KB stream live instead). A body whose frames were lost in transit therefore never reaches the loopback server as an empty/truncated chunked body — the host aborts the stream and the client sees an ambiguous transport failure it can retry, instead of the loopback server's bare `400` ("Failed to send message (400)" from the mobile app).
+
 ## Authentication model

 - The tunnel is **transport only**. The OpenChamber server still authenticates every tunneled request exactly as it authenticates a direct remote client. The relay path grants reachability, not authorization.
@@ -62,6 +62,17 @@ const STRIPPED_RESPONSE_HEADERS = new Set([
 const BACKPRESSURE_LIMIT_BYTES = 4 * 1024 * 1024;
 const BACKPRESSURE_POLL_MS = 20;

+// Bodies smaller than this are fully buffered before the loopback request is
+// sent, so a tunneled body that lost frames (relay reconnect, dropped HttpBody
+// frames) can never reach the loopback server as an empty/truncated chunked
+// body — the server rejects those with a bare 400, surfacing as the mobile
+// app's "Failed to send message (400)". Larger bodies stream live as before.
+const BODY_BUFFER_MAX_BYTES = 512 * 1024;
+// While the body is still being buffered, abort the stream if it never
+// completes, so a stalled tunnel converts into an ambiguous transport failure
+// (which the client already retries) instead of a hung loopback request.
+const BODY_DELIVERY_TIMEOUT_MS = 15_000;
+
 const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

 const isHttpRequestPayload = (parsed) =>
@@ -168,39 +179,14 @@ export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBuf
    await send(encodeTunnelFrame(TunnelFrameType.StreamEnd, streamId, new Uint8Array(0)));
  };

-  const runHttpStream = async (streamId, request) => {
-    const method = request.method.toUpperCase();
-    if (!isAllowedHttpPath(request.path)) {
-      dropStream(streamId);
-      await syntheticResponse(streamId, 403, 'Path is not allowed through the relay');
-      return;
-    }
-
-    const stream = streams.get(streamId);
-    if (!stream || stream.kind !== 'http') return;
-
-    const hasBody = method !== 'GET' && method !== 'HEAD';
-    let requestBody;
-    if (hasBody) {
-      requestBody = new ReadableStream({
-        start(controller) {
-          stream.body = controller;
-        },
-      });
-    } else {
-      stream.body = null;
-      stream.noBody = true;
-    }
-
-    const loopbackOrigin = `http://127.0.0.1:${getLocalPort()}`;
-    const url = `${loopbackOrigin}${request.path}${request.query ? `?${request.query}` : ''}`;
+  const forwardRequest = async (streamId, stream, url, method, request, body, loopbackOrigin) => {
    let response;
    try {
      response = await fetch(url, {
        method,
        headers: buildRequestHeaders(request.headers, loopbackOrigin),
-        body: requestBody,
-        duplex: hasBody ? 'half' : undefined,
+        body,
+        duplex: body ? 'half' : undefined,
        signal: stream.abort.signal,
      });
    } catch (error) {
@@ -242,6 +228,139 @@ export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBuf
    }
  };

+  const runHttpStream = async (streamId, request) => {
+    const method = request.method.toUpperCase();
+    if (!isAllowedHttpPath(request.path)) {
+      dropStream(streamId);
+      await syntheticResponse(streamId, 403, 'Path is not allowed through the relay');
+      return;
+    }
+
+    const stream = streams.get(streamId);
+    if (!stream || stream.kind !== 'http') return;
+
+    const hasBody = method !== 'GET' && method !== 'HEAD';
+    const loopbackOrigin = `http://127.0.0.1:${getLocalPort()}`;
+    const url = `${loopbackOrigin}${request.path}${request.query ? `?${request.query}` : ''}`;
+
+    if (!hasBody) {
+      stream.noBody = true;
+      await forwardRequest(streamId, stream, url, method, request, null, loopbackOrigin);
+      return;
+    }
+
+    // Body-carrying request. Buffer the tunneled body frames and forward the
+    // COMPLETE body only once StreamEnd arrives. Forwarding a body that lost
+    // frames through the tunnel (relay reconnect, dropped HttpBody frames)
+    // reaches the loopback server as an empty/truncated chunked body, which it
+    // rejects with a bare 400 (empty response body) — the "Failed to send
+    // message (400)" seen from the mobile APK. Bodies above BODY_BUFFER_MAX_BYTES
+    // stream live so large uploads are not fully buffered.
+    const buffered = [];
+    let bufferedBytes = 0;
+    let bodyFrameCount = 0;
+    let liveStream = null;
+    let liveController = null;
+    let completed = false;
+    let bodyFailure = null;
+    let resolveBodyEnd;
+    const bodyEnded = new Promise((resolve) => { resolveBodyEnd = resolve; });
+
+    const finishBody = (error) => {
+      if (completed) return;
+      completed = true;
+      bodyFailure = error ?? null;
+      if (liveController) {
+        try {
+          if (error) liveController.error(error);
+          else liveController.close();
+        } catch {
+          // stream already errored/closed
+        }
+      }
+      resolveBodyEnd();
+    };
+
+    let deliveryDeadline = null;
+    const switchToLive = () => {
+      liveStream = new ReadableStream({
+        start(controller) {
+          liveController = controller;
+          stream.body = controller;
+        },
+      });
+      for (const chunk of buffered) {
+        try { liveController.enqueue(chunk); } catch { break; }
+      }
+      buffered.length = 0;
+      // The loopback request is now streaming live; runHttpStream has nothing
+      // left to do — clear the deadline and let the async body forwarding own
+      // this stream from here (abort/StreamEnd close the controller).
+      if (deliveryDeadline) clearTimeout(deliveryDeadline);
+      resolveBodyEnd();
+      void forwardRequest(streamId, stream, url, method, request, liveStream, loopbackOrigin);
+    };
+
+    stream.body = {
+      enqueue(payload) {
+        if (completed) return;
+        bodyFrameCount += 1;
+        if (liveController) {
+          try { liveController.enqueue(payload); } catch {
+            // stream already errored/closed
+          }
+          return;
+        }
+        buffered.push(payload);
+        bufferedBytes += payload.length;
+        if (bufferedBytes > BODY_BUFFER_MAX_BYTES) {
+          switchToLive();
+        }
+      },
+      close() {
+        finishBody(null);
+      },
+      error(error) {
+        finishBody(error);
+      },
+    };
+
+    deliveryDeadline = setTimeout(() => {
+      if (streams.get(streamId) === stream && !completed && !liveStream) {
+        dropStream(streamId);
+        void sendAbort(streamId, 'tunnel request body was not delivered in time');
+      }
+    }, BODY_DELIVERY_TIMEOUT_MS);
+    deliveryDeadline.unref?.();
+
+    await bodyEnded;
+    if (deliveryDeadline) clearTimeout(deliveryDeadline);
+    if (streams.get(streamId) !== stream) return; // aborted or dropped meanwhile
+    if (bodyFailure) {
+      dropStream(streamId);
+      await sendAbort(streamId, bodyFailure.message ?? 'tunnel request body failed');
+      return;
+    }
+    if (liveStream) return; // already forwarded via the streaming path
+
+    // The client signaled it had a body but no HttpBody frame arrived before
+    // StreamEnd — the body frames were lost through the tunnel. Forwarding an
+    // empty body would make the loopback server reject the request with a bare
+    // 400. Abort instead so the client treats it as an ambiguous transport
+    // failure (dispatched, outcome unknown) and can safely retry.
+    if (request.hasBody === true && bodyFrameCount === 0) {
+      dropStream(streamId);
+      await sendAbort(streamId, 'tunnel request body frames were lost');
+      return;
+    }
+
+    // Buffered path: forward the complete body as a single buffer so Bun
+    // frames it with content-length — never as a chunked body that could be
+    // truncated. Reset the buffered handler so late frames cannot enqueue.
+    stream.body = null;
+    await forwardRequest(streamId, stream, url, method, request, Buffer.concat(buffered), loopbackOrigin);
+  };
+
  const handleHttpRequest = (streamId, payload) => {
    if (streams.has(streamId)) {
      abortLocalStream(streamId, 'duplicate stream id');
@@ -263,9 +382,9 @@ export const createTunnelHost = ({ connectionId, getLocalPort, sendFrame, getBuf
  const handleHttpBody = (streamId, payload) => {
    const stream = streams.get(streamId);
    if (!stream || stream.kind !== 'http' || stream.noBody) return;
-    // The body controller attaches synchronously in runHttpStream before any
-    // await, so by the time body frames arrive it is set for body-carrying
-    // methods; drop stray body bytes otherwise.
+    // runHttpStream installs a body sink (buffering handler, or the live stream
+    // controller once the buffer cap is crossed) before any HttpBody frame can
+    // arrive; drop stray bytes for request bodies already completed/aborted.
    try {
      stream.body?.enqueue(payload);
    } catch {
@@ -0,0 +1,116 @@
+import { describe, test, expect } from 'bun:test';
+import http from 'node:http';
+
+import { createTunnelHost } from './tunnel-host.js';
+import { decodeTunnelFrame, encodeTunnelFrame, encodeJsonPayload, TunnelFrameType } from './tunnel-codec.js';
+
+const startLoopback = () =>
+  new Promise((resolve) => {
+    const requests = [];
+    const server = http.createServer((req, res) => {
+      const chunks = [];
+      req.on('data', (c) => chunks.push(c));
+      req.on('end', () => {
+        requests.push({
+          method: req.method,
+          url: req.url,
+          body: Buffer.concat(chunks).toString('utf8'),
+        });
+        res.writeHead(200, { 'content-type': 'application/json' });
+        res.end(JSON.stringify({ ok: true }));
+      });
+    });
+    server.listen(0, '127.0.0.1', () => resolve({
+      server,
+      port: server.address().port,
+      requests,
+      stop: () => new Promise((r) => server.close(() => r())),
+    }));
+  });
+
+const createHarness = async () => {
+  const loopback = await startLoopback();
+  const sentFrames = [];
+  const host = createTunnelHost({
+    connectionId: 'conn-test',
+    getLocalPort: () => loopback.port,
+    sendFrame: async (frame) => {
+      sentFrames.push(decodeTunnelFrame(frame));
+    },
+    getBufferedAmount: () => 0,
+  });
+  return { host, loopback, sentFrames };
+};
+
+const httpHead = (overrides = {}) => encodeTunnelFrame(TunnelFrameType.HttpRequest, 1, encodeJsonPayload({
+  method: 'POST',
+  path: '/api/submit',
+  query: '',
+  headers: { 'content-type': 'application/json' },
+  ...overrides,
+}));
+
+const waitFor = async (predicate, timeoutMs = 2000) => {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    if (predicate()) return true;
+    await new Promise((r) => setTimeout(r, 20));
+  }
+  return predicate();
+};
+
+describe('tunnel-host HTTP body forwarding', () => {
+  test('buffers tunneled body frames and forwards the complete body', async () => {
+    const { host, loopback, sentFrames } = await createHarness();
+    await host.handleFrame(httpHead());
+    await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpBody, 1, new TextEncoder().encode('alpha')));
+    await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpBody, 1, new TextEncoder().encode('beta')));
+    await host.handleFrame(encodeTunnelFrame(TunnelFrameType.StreamEnd, 1, new Uint8Array(0)));
+
+    const received = await waitFor(() => loopback.requests.length === 1);
+    expect(received).toBe(true);
+    expect(loopback.requests[0].method).toBe('POST');
+    expect(loopback.requests[0].body).toBe('alphabeta');
+    await waitFor(() => sentFrames.some((f) => f.frameType === TunnelFrameType.StreamEnd));
+    await loopback.stop();
+  });
+
+  test('body-expected request with zero delivered frames is aborted as ambiguous, not forwarded', async () => {
+    const { host, loopback, sentFrames } = await createHarness();
+    await host.handleFrame(httpHead({ hasBody: true }));
+    await host.handleFrame(encodeTunnelFrame(TunnelFrameType.StreamEnd, 1, new Uint8Array(0)));
+
+    const aborted = await waitFor(() => sentFrames.some((f) => f.frameType === TunnelFrameType.StreamAbort));
+    expect(aborted).toBe(true);
+    // Loopback must never have seen a request with a lost body.
+    expect(loopback.requests.length).toBe(0);
+    await loopback.stop();
+  });
+
+  test('bodyless request (hasBody absent, legacy client) still forwards empty', async () => {
+    const { host, loopback } = await createHarness();
+    await host.handleFrame(httpHead());
+    await host.handleFrame(encodeTunnelFrame(TunnelFrameType.StreamEnd, 1, new Uint8Array(0)));
+
+    const received = await waitFor(() => loopback.requests.length === 1);
+    expect(received).toBe(true);
+    expect(loopback.requests[0].body).toBe('');
+    await loopback.stop();
+  });
+
+  test('GET forwards immediately with no body wait', async () => {
+    const { host, loopback } = await createHarness();
+    await host.handleFrame(encodeTunnelFrame(TunnelFrameType.HttpRequest, 1, encodeJsonPayload({
+      method: 'GET',
+      path: '/api/health',
+      query: '',
+      headers: {},
+    })));
+    await host.handleFrame(encodeTunnelFrame(TunnelFrameType.StreamEnd, 1, new Uint8Array(0)));
+
+    const received = await waitFor(() => loopback.requests.length === 1);
+    expect(received).toBe(true);
+    expect(loopback.requests[0].method).toBe('GET');
+    await loopback.stop();
+  });
+});