From 97b520e91641a402269cf7d470c8f2596f2b53ad Mon Sep 17 00:00:00 2001 From: jschaul Date: Thu, 23 Jul 2026 19:00:43 +0200 Subject: [PATCH 01/10] another attempt at solving the absurdity --- services/federator/src/Federator/Response.hs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/services/federator/src/Federator/Response.hs b/services/federator/src/Federator/Response.hs index 7633252c287..a55d9864fda 100644 --- a/services/federator/src/Federator/Response.hs +++ b/services/federator/src/Federator/Response.hs @@ -25,7 +25,17 @@ import Servant.Types.SourceT streamingResponseToWai :: StreamingResponse -> Wai.Response streamingResponseToWai resp = - let headers = toList (responseHeaders resp) + let -- We re-frame the body ourselves via 'Wai.responseStream' (Warp emits it + -- chunked), so any framing header from the upstream response must be + -- dropped. Passing a stale 'Content-Length' or 'Transfer-Encoding' + -- through would double-frame the body: the peer reads the advertised + -- length, treats the response as complete, and leaves the remaining bytes + -- in the socket buffer. On a reused keep-alive connection the next + -- request then reads those leftover bytes as its own response, which + -- surfaces as an unrelated 200 with a non-JSON body (e.g. a buffered + -- /i/metrics scrape) and manifests as flaky federation calls. + isFramingHeader (name, _) = name == "Content-Length" || name == "Transfer-Encoding" + headers = filter (not . isFramingHeader) (toList (responseHeaders resp)) status = responseStatusCode resp streamingBody output flush = foreach From b829a0ce301d30000521aca4e6680e093a117c43 Mon Sep 17 00:00:00 2001 From: jschaul Date: Thu, 23 Jul 2026 21:49:06 +0200 Subject: [PATCH 02/10] Revert "another attempt at solving the absurdity" This reverts commit 97b520e91641a402269cf7d470c8f2596f2b53ad. --- services/federator/src/Federator/Response.hs | 12 +----------- 1 file changed, 1 insertion(+), 11 deletions(-) diff --git a/services/federator/src/Federator/Response.hs b/services/federator/src/Federator/Response.hs index a55d9864fda..7633252c287 100644 --- a/services/federator/src/Federator/Response.hs +++ b/services/federator/src/Federator/Response.hs @@ -25,17 +25,7 @@ import Servant.Types.SourceT streamingResponseToWai :: StreamingResponse -> Wai.Response streamingResponseToWai resp = - let -- We re-frame the body ourselves via 'Wai.responseStream' (Warp emits it - -- chunked), so any framing header from the upstream response must be - -- dropped. Passing a stale 'Content-Length' or 'Transfer-Encoding' - -- through would double-frame the body: the peer reads the advertised - -- length, treats the response as complete, and leaves the remaining bytes - -- in the socket buffer. On a reused keep-alive connection the next - -- request then reads those leftover bytes as its own response, which - -- surfaces as an unrelated 200 with a non-JSON body (e.g. a buffered - -- /i/metrics scrape) and manifests as flaky federation calls. - isFramingHeader (name, _) = name == "Content-Length" || name == "Transfer-Encoding" - headers = filter (not . isFramingHeader) (toList (responseHeaders resp)) + let headers = toList (responseHeaders resp) status = responseStatusCode resp streamingBody output flush = foreach From e3d1373f7af0b67662ca0ff0c23a67290330e3b3 Mon Sep 17 00:00:00 2001 From: jschaul Date: Thu, 23 Jul 2026 22:02:53 +0200 Subject: [PATCH 03/10] ... --- services/federator/federator.cabal | 4 + .../federator/src/Federator/InternalServer.hs | 19 + services/federator/test/unit/Main.hs | 4 +- .../test/unit/Test/Federator/Response.hs | 357 ++++++++++++++++++ 4 files changed, 383 insertions(+), 1 deletion(-) create mode 100644 services/federator/test/unit/Test/Federator/Response.hs diff --git a/services/federator/federator.cabal b/services/federator/federator.cabal index 872f9911236..6668d84e36c 100644 --- a/services/federator/federator.cabal +++ b/services/federator/federator.cabal @@ -326,6 +326,7 @@ test-suite federator-tests Test.Federator.Monitor Test.Federator.Options Test.Federator.Remote + Test.Federator.Response Test.Federator.Util Test.Federator.Validation @@ -380,6 +381,7 @@ test-suite federator-tests build-depends: aeson + , async , base , bytestring , bytestring-conversion @@ -391,6 +393,7 @@ test-suite federator-tests , federator , filepath , HsOpenSSL + , http-client , http-types , http2 , http2-manager @@ -398,6 +401,7 @@ test-suite federator-tests , interpolate , kan-extensions , mtl + , network , polysemy , polysemy-wire-zoo , QuickCheck diff --git a/services/federator/src/Federator/InternalServer.hs b/services/federator/src/Federator/InternalServer.hs index accd160987d..288a1504b57 100644 --- a/services/federator/src/Federator/InternalServer.hs +++ b/services/federator/src/Federator/InternalServer.hs @@ -43,6 +43,7 @@ import Polysemy.TinyLog import Servant qualified import Servant.API import Servant.API.Extended.Endpath +import Servant.Client.Core (responseHeaders, responseStatusCode) import Servant.Server.Generic import System.Logger.Class qualified as Log import Wire.API.Federation.Component @@ -127,6 +128,24 @@ callOutward targetDomain component (RPC path) req cont = do path (Wai.requestHeaders req) (fromLazyByteString body) + -- Diagnostic for a known flaky failure mode (WPB: federation ingress probe + -- occasionally receives a 200 with a non-JSON body). The federator is a + -- transparent byte proxy, so a federation RPC coming back as a 200 whose body + -- is not JSON means the outward call reached something other than the target's + -- federation endpoint (e.g. an ingress/route not yet programmed during backend + -- startup, or a stale/mismatched response on a reused connection). We only + -- inspect the status line and Content-Type header here; the body must not be + -- consumed, as it is streamed straight back to the caller. + let respStatus = HTTP.statusCode (responseStatusCode resp) + respContentType = snd <$> find ((== HTTP.hContentType) . fst) (responseHeaders resp) + when (respStatus == 200 && respContentType /= Just "application/json") $ + warn $ + Log.msg (Log.val "Federator outward call returned a non-JSON 200; upstream is not the target's federation endpoint") + . Log.field "domain" targetDomain._domainText + . Log.field "component" (show component) + . Log.field "path" path + . Log.field "status" (show respStatus) + . Log.field "contentType" (maybe ("" :: ByteString) id respContentType) embed . cont $ streamingResponseToWai resp serveOutward :: Env -> Int -> IORef [IO ()] -> IO () diff --git a/services/federator/test/unit/Main.hs b/services/federator/test/unit/Main.hs index 1936df48d0d..f58f7a80652 100644 --- a/services/federator/test/unit/Main.hs +++ b/services/federator/test/unit/Main.hs @@ -28,6 +28,7 @@ import Test.Federator.InternalServer qualified import Test.Federator.Monitor qualified import Test.Federator.Options qualified import Test.Federator.Remote qualified +import Test.Federator.Response qualified import Test.Federator.Validation qualified import Test.Tasty @@ -43,5 +44,6 @@ main = Test.Federator.InternalServer.tests, Test.Federator.ExternalServer.tests, Test.Federator.Monitor.tests, - Test.Federator.Remote.tests + Test.Federator.Remote.tests, + Test.Federator.Response.tests ] diff --git a/services/federator/test/unit/Test/Federator/Response.hs b/services/federator/test/unit/Test/Federator/Response.hs new file mode 100644 index 00000000000..4a394c52083 --- /dev/null +++ b/services/federator/test/unit/Test/Federator/Response.hs @@ -0,0 +1,357 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2022 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) any +-- later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +-- | Local reproduction harness for the "federation RPC comes back as a 200 with +-- a non-JSON body" flake. +-- +-- The federation @/rpc@ and @/federation@ endpoints are served over the same +-- keep-alive HTTP/1.1 port as @/i/metrics@. The observed failure is a @POST +-- /rpc/…@ returning a 200 whose body is an unrelated @/i/metrics@ page — i.e. a +-- response the connection should never have produced for that request. That only +-- happens if a response mis-frames the connection (leaving bytes behind) so a +-- later request on the reused connection reads them. +-- +-- Two angles are exercised here: +-- +-- 1. 'testEmptyChunkDoesNotTruncate' drives the ACTUAL +-- 'Federator.Response.streamingResponseToWai' with an upstream body that +-- yields an empty chunk in the middle of the stream (which a streamed HTTP/2 +-- upstream legitimately can). If Warp turned that into a premature chunked +-- terminator, the response body would be truncated and the next pipelined +-- request would desync. This pins down whether the streaming path can poison +-- a keep-alive connection. +-- +-- 2. 'testConcurrentSharedManagerStaysInSync' stands up a federator-mimic (a +-- chunked @/i/metrics@ and an @/rpc@ served through 'streamingResponseToWai') +-- and hammers it through a SINGLE shared 'http-client' 'Manager' with a small +-- connection pool — the same setup the integration suite uses — mixing metric +-- scrapes and RPCs concurrently. It asserts no RPC response is ever poisoned +-- by connection reuse. It is a control: if well-formed responses never +-- desync here, the trigger must be a genuine mis-framing (see angle 1) rather +-- than the shared Manager alone. +module Test.Federator.Response (tests) where + +import Control.Concurrent.Async (forConcurrently) +import Control.Exception (bracket) +import Data.Bifunctor (first) +import Data.ByteString qualified as BS +import Data.ByteString.Builder (byteString) +import Data.ByteString.Char8 qualified as BS8 +import Data.ByteString.Lazy qualified as L +import Data.Sequence qualified as Seq +import Federator.Response (streamingResponseToWai) +import Imports +import Network.HTTP.Client qualified as HTTP +import Network.HTTP.Types qualified as HTTP +import Network.Socket +import Network.Socket.ByteString (recv, sendAll) +import Network.Wai qualified as Wai +import Network.Wai.Handler.Warp qualified as Warp +import Numeric (readHex) +import Servant.Client.Core (ResponseF (..), StreamingResponse) +import Servant.Types.SourceT (source) +import System.Timeout (timeout) +import Test.Tasty +import Test.Tasty.HUnit + +tests :: TestTree +tests = + testGroup + "Response.streamingResponseToWai / keep-alive framing" + [ testCase + "empty upstream chunk does not truncate the streamed response or desync the connection" + testEmptyChunkDoesNotTruncate, + testCase + "shared http-client Manager under concurrent metrics+RPC never delivers a poisoned RPC response" + testConcurrentSharedManagerStaysInSync + ] + +-- | A genuine JSON federation response body (what @/brig/api-version@ returns). +realFederationBody :: ByteString +realFederationBody = "{\"supportedVersions\":[0,1,2]}" + +-- | Build an upstream streaming response (as produced by the outward HTTP/2 +-- call) carrying only a Content-Type header, so 'streamingResponseToWai' streams +-- it chunked, exactly like the real federation path. +jsonUpstream :: [ByteString] -> StreamingResponse +jsonUpstream chunks = + Response + { responseStatusCode = HTTP.ok200, + responseHeaders = Seq.fromList [(HTTP.hContentType, "application/json")], + responseHttpVersion = HTTP.http20, + responseBody = source chunks + } + +-------------------------------------------------------------------------------- +-- Angle 1: empty chunk in the middle of a streamed body. + +-- | The upstream yields ["{\"a\":", "", "1}"] — an empty chunk between two +-- non-empty ones. A conforming server must still deliver the concatenation +-- ("{\"a\":1}") and keep the connection framed correctly. +emptyChunkUpstream :: StreamingResponse +emptyChunkUpstream = jsonUpstream ["{\"a\":", "", "1}"] + +testEmptyChunkDoesNotTruncate :: Assertion +testEmptyChunkDoesNotTruncate = + Warp.testWithApplication (pure (mkApp (streamingResponseToWai emptyChunkUpstream))) $ \port -> do + raw <- singleRequestRaw port "/first" + (r1, r2) <- pipelineTwo port + putStrLn $ + unlines + [ "", + "===== empty-chunk framing =====", + "raw bytes of GET /first: " <> show raw, + "parsed 1st response: " <> show r1, + "parsed 2nd response: " <> show r2 + ] + assertEqual "first response body must be the full concatenation of all chunks" "{\"a\":1}" (hrBody r1) + assertEqual "second (pipelined) response must be intact (connection not desynced)" legitSecond r2 + +-------------------------------------------------------------------------------- +-- Angle 2: shared Manager under concurrency (control). + +-- | A metrics page big enough to span several chunks, framed chunked (no +-- Content-Length) just like the prometheus middleware serves @/i/metrics@. +metricsBody :: ByteString +metricsBody = + "# HELP http_request_duration_seconds The HTTP request latencies in seconds.\n" + <> "# TYPE http_request_duration_seconds histogram\n" + <> mconcat + [ "http_request_duration_seconds_bucket{handler=\"/rpc\",le=\"" <> BS8.pack (show (i :: Int)) <> "\"} 1\n" + | i <- [1 .. 400] + ] + +federatorMimicApp :: Wai.Application +federatorMimicApp req respond = + case Wai.rawPathInfo req of + "/i/metrics" -> + respond $ + Wai.responseBuilder + HTTP.ok200 + [(HTTP.hContentType, "text/plain; version=0.0.4")] + (byteString metricsBody) + _ -> respond (streamingResponseToWai (jsonUpstream [realFederationBody])) + +testConcurrentSharedManagerStaysInSync :: Assertion +testConcurrentSharedManagerStaysInSync = + Warp.testWithApplication (pure federatorMimicApp) $ \port -> do + -- One shared Manager with a small pool, so RPC and metric requests are forced + -- to reuse the same handful of keep-alive connections (as in the suite). + mgr <- HTTP.newManager HTTP.defaultManagerSettings {HTTP.managerConnCount = 4} + let n = 400 :: Int + base = "http://127.0.0.1:" <> show port + oneRequest i + | even i = do + req <- HTTP.parseRequest ("GET " <> base <> "/i/metrics") + _ <- HTTP.httpLbs req mgr + pure Nothing + | otherwise = do + req0 <- HTTP.parseRequest (base <> "/rpc/d.example.com/brig/api-version") + resp <- HTTP.httpLbs req0 {HTTP.method = "POST"} mgr + let body = L.toStrict (HTTP.responseBody resp) + ct = lookup HTTP.hContentType (HTTP.responseHeaders resp) + status = HTTP.statusCode (HTTP.responseStatus resp) + pure $ + if status == 200 && ct == Just "application/json" && body == realFederationBody + then Nothing + else Just (i, status, ct, body) + poisoned <- catMaybes <$> forConcurrently [1 .. n] oneRequest + putStrLn $ + "\n===== shared-Manager concurrency =====" + <> "\nRPC calls: " + <> show (length (filter odd [1 .. n])) + <> ", metric scrapes: " + <> show (length (filter even [1 .. n])) + <> ", poisoned RPC responses: " + <> show (length poisoned) + for_ poisoned $ \p -> putStrLn (" POISONED: " <> show p) + assertBool + ("expected no RPC response to be poisoned by connection reuse, but got: " <> show poisoned) + (null poisoned) + +-------------------------------------------------------------------------------- +-- Test server plumbing. + +-- | Serve the given (already-rendered) response on @/first@; a distinct, plain +-- JSON response on anything else (the "next request" whose integrity we check). +mkApp :: Wai.Response -> Wai.Application +mkApp firstResp req respond = + case Wai.rawPathInfo req of + "/first" -> respond firstResp + _ -> + respond $ + Wai.responseLBS + HTTP.ok200 + [(HTTP.hContentType, "application/json")] + "{\"second\":true}" + +legitSecond :: HttpResponse +legitSecond = + HttpResponse + { hrStatus = 200, + hrContentType = Just "application/json", + hrBody = "{\"second\":true}" + } + +-------------------------------------------------------------------------------- +-- A minimal, conforming HTTP/1.1 client over a raw socket. It follows exactly +-- the framing the server advertises (Content-Length, else chunked), which is +-- where a mis-framed response bites a reused connection. + +data HttpResponse = HttpResponse + { hrStatus :: Int, + hrContentType :: Maybe ByteString, + hrBody :: ByteString + } + deriving (Eq, Show) + +-- | Pipeline two requests on one keep-alive connection, then parse both +-- responses in order (deterministic byte interleaving). +pipelineTwo :: Int -> IO (HttpResponse, HttpResponse) +pipelineTwo port = do + addr <- resolve port + bracket (open addr) close $ \sock -> do + buf <- newBuf sock + sendAll sock $ + "GET /first HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n" + <> "GET /second HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n" + r1 <- parseResponse buf + r2 <- parseResponse buf + pure (r1, r2) + +-- | Single request on a fresh connection; slurp bytes until the socket is idle +-- (keep-alive means no EOF). Evidence only. +singleRequestRaw :: Int -> ByteString -> IO ByteString +singleRequestRaw port path = do + addr <- resolve port + bracket (open addr) close $ \sock -> do + sendAll sock $ "GET " <> path <> " HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n" + slurp sock mempty + where + slurp sock acc = do + mchunk <- timeout 400_000 (recv sock 4096) + case mchunk of + Just chunk | not (BS.null chunk) -> slurp sock (acc <> chunk) + _ -> pure acc + +data Buf = Buf Socket (IORef ByteString) + +newBuf :: Socket -> IO Buf +newBuf s = Buf s <$> newIORef mempty + +fill :: Buf -> IO Bool +fill (Buf s ref) = do + more <- recv s 4096 + if BS.null more then pure False else True <$ modifyIORef' ref (<> more) + +readLineB :: Buf -> IO ByteString +readLineB buf@(Buf _ ref) = go + where + go = do + b <- readIORef ref + case breakOnCRLF b of + Just (line, rest) -> line <$ writeIORef ref rest + Nothing -> do + ok <- fill buf + if ok then go else b <$ writeIORef ref mempty + +readNB :: Buf -> Int -> IO ByteString +readNB buf@(Buf _ ref) n = go + where + go = do + b <- readIORef ref + if BS.length b >= n + then let (h, t) = BS.splitAt n b in h <$ writeIORef ref t + else do + ok <- fill buf + if ok then go else b <$ writeIORef ref mempty + +breakOnCRLF :: ByteString -> Maybe (ByteString, ByteString) +breakOnCRLF b = + case BS.breakSubstring "\r\n" b of + (pre, rest) + | BS.null rest -> Nothing + | otherwise -> Just (pre, BS.drop 2 rest) + +parseResponse :: Buf -> IO HttpResponse +parseResponse buf = do + statusLine <- readLineB buf + hdrs <- readHeaders buf [] + body <- readBody buf hdrs + pure + HttpResponse + { hrStatus = parseStatus statusLine, + hrContentType = lookupCI "content-type" hdrs, + hrBody = body + } + where + parseStatus l + | "HTTP/" `BS.isPrefixOf` l = + case BS8.words l of + (_ver : codeBS : _) -> fromMaybe (-1) (readMaybe (BS8.unpack codeBS)) + _ -> -1 + | otherwise = -1 + +readHeaders :: Buf -> [(ByteString, ByteString)] -> IO [(ByteString, ByteString)] +readHeaders buf acc = do + line <- readLineB buf + if BS.null line + then pure (reverse acc) + else readHeaders buf (splitHeader line : acc) + where + splitHeader line = + case BS.breakSubstring ": " line of + (k, rest) + | BS.null rest -> (line, "") + | otherwise -> (k, BS.drop 2 rest) + +readBody :: Buf -> [(ByteString, ByteString)] -> IO ByteString +readBody buf hdrs = + case lookupCI "content-length" hdrs of + Just clBS | Just n <- readMaybe (BS8.unpack clBS) -> readNB buf n + _ -> case lookupCI "transfer-encoding" hdrs of + Just te | "chunked" `BS.isInfixOf` BS8.map toLower te -> readChunked buf mempty + _ -> pure mempty + +readChunked :: Buf -> ByteString -> IO ByteString +readChunked buf acc = do + sizeLine <- readLineB buf + let sizeHex = BS8.takeWhile (/= ';') sizeLine + case readHex (BS8.unpack sizeHex) of + [(0 :: Int, _)] -> acc <$ readLineB buf -- consume trailing CRLF after last chunk + [(n, _)] -> do + chunk <- readNB buf n + _ <- readNB buf 2 -- trailing CRLF after the chunk + readChunked buf (acc <> chunk) + _ -> pure acc + +lookupCI :: ByteString -> [(ByteString, ByteString)] -> Maybe ByteString +lookupCI key = lookup key . map (first (BS8.map toLower)) + +resolve :: Int -> IO AddrInfo +resolve port = + head + <$> getAddrInfo + (Just defaultHints {addrSocketType = Stream}) + (Just "127.0.0.1") + (Just (show port)) + +open :: AddrInfo -> IO Socket +open addr = do + sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr) + connect sock (addrAddress addr) + pure sock From 745e0783665cf3e8c98b4d07bc4be5eebf729666 Mon Sep 17 00:00:00 2001 From: jschaul Date: Thu, 23 Jul 2026 22:25:41 +0200 Subject: [PATCH 04/10] lint --- services/federator/default.nix | 3 +++ services/federator/src/Federator/InternalServer.hs | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/services/federator/default.nix b/services/federator/default.nix index fa8eac96bda..25db4b88eaa 100644 --- a/services/federator/default.nix +++ b/services/federator/default.nix @@ -168,6 +168,7 @@ mkDerivation { ]; testHaskellDepends = [ aeson + async base bytestring bytestring-conversion @@ -178,6 +179,7 @@ mkDerivation { dns-util filepath HsOpenSSL + http-client http-types http2 http2-manager @@ -185,6 +187,7 @@ mkDerivation { interpolate kan-extensions mtl + network polysemy polysemy-wire-zoo QuickCheck diff --git a/services/federator/src/Federator/InternalServer.hs b/services/federator/src/Federator/InternalServer.hs index 288a1504b57..2dfaee4e9c7 100644 --- a/services/federator/src/Federator/InternalServer.hs +++ b/services/federator/src/Federator/InternalServer.hs @@ -145,7 +145,7 @@ callOutward targetDomain component (RPC path) req cont = do . Log.field "component" (show component) . Log.field "path" path . Log.field "status" (show respStatus) - . Log.field "contentType" (maybe ("" :: ByteString) id respContentType) + . Log.field "contentType" (fromMaybe ("" :: ByteString) respContentType) embed . cont $ streamingResponseToWai resp serveOutward :: Env -> Int -> IORef [IO ()] -> IO () From 9f24a7e338ab1e64231eb21a1197febddbf57e6c Mon Sep 17 00:00:00 2001 From: jschaul Date: Thu, 23 Jul 2026 22:36:16 +0200 Subject: [PATCH 05/10] Hi CI From 266ca4385be9472c445eb4fed96e09e79f5f944c Mon Sep 17 00:00:00 2001 From: jschaul Date: Thu, 23 Jul 2026 22:49:15 +0200 Subject: [PATCH 06/10] Hi CI From e1df2531a170c35d1c32a7a2ae4be0e2e78dbb15 Mon Sep 17 00:00:00 2001 From: jschaul Date: Wed, 29 Jul 2026 11:26:04 +0200 Subject: [PATCH 07/10] deflake-connection-reuse --- .../federator-streaming-response-framing | 1 + services/federator/src/Federator/Response.hs | 23 ++- .../test/unit/Test/Federator/Response.hs | 190 +++++++++++------- 3 files changed, 138 insertions(+), 76 deletions(-) create mode 100644 changelog.d/3-bug-fixes/federator-streaming-response-framing diff --git a/changelog.d/3-bug-fixes/federator-streaming-response-framing b/changelog.d/3-bug-fixes/federator-streaming-response-framing new file mode 100644 index 00000000000..ae6fbdd737e --- /dev/null +++ b/changelog.d/3-bug-fixes/federator-streaming-response-framing @@ -0,0 +1 @@ +Federator: stop forwarding the upstream `Content-Length`/`Transfer-Encoding` headers when re-streaming a federation response. Warp re-frames the streamed body itself (chunked); forwarding a `Content-Length` made Warp serve the body under that declared length instead, so a truncated/reset cold-start upstream (declared length ≠ streamed bytes) desynchronised the caller's keep-alive connection and could deliver an unrelated response (e.g. a `/i/metrics` page in reply to a `POST /rpc/.../brig/api-version`). This was the cause of the flaky `Federation.testNotificationsForOfflineBackends` / `Federator.testFederatorNumRequestsMetrics` failures. diff --git a/services/federator/src/Federator/Response.hs b/services/federator/src/Federator/Response.hs index 7633252c287..4226d5dffef 100644 --- a/services/federator/src/Federator/Response.hs +++ b/services/federator/src/Federator/Response.hs @@ -19,13 +19,32 @@ module Federator.Response where import Data.ByteString.Builder import Imports +import Network.HTTP.Types.Header (hContentLength, hTransferEncoding) import Network.Wai qualified as Wai import Servant.Client.Core import Servant.Types.SourceT +-- | Turn a streaming upstream response (from the outward federation call) into a +-- WAI response that Warp serves back to the caller. +-- +-- We re-frame the body: Warp streams it with chunked transfer-encoding (it does +-- not know the length up front). We must therefore DROP the upstream's own +-- framing headers ('Content-Length', 'Transfer-Encoding'); forwarding them is a +-- keep-alive desync waiting to happen. +-- +-- In particular, if we forward a 'Content-Length', Warp honours it verbatim and +-- sends the streamed body raw under that declared length instead of chunking it. +-- The moment the declared length disagrees with the number of bytes we actually +-- stream — a truncated or reset cold-start upstream, a stale @Content-Length@ — +-- the client reads exactly the declared number of bytes and runs straight past +-- the response boundary into the next response on the reused connection. On the +-- integration suite's shared, pooled HTTP/1.1 connection that surfaces as a +-- @POST /rpc/…@ coming back with an unrelated @/i/metrics@ body. Stripping the +-- framing headers lets Warp frame exactly what we stream, so the length on the +-- wire can never disagree with the body and the connection stays in sync. streamingResponseToWai :: StreamingResponse -> Wai.Response streamingResponseToWai resp = - let headers = toList (responseHeaders resp) + let headers = filter (not . isFramingHeader . fst) (toList (responseHeaders resp)) status = responseStatusCode resp streamingBody output flush = foreach @@ -33,3 +52,5 @@ streamingResponseToWai resp = (\chunk -> output (byteString chunk) *> flush) (responseBody resp) in Wai.responseStream status headers streamingBody + where + isFramingHeader h = h == hContentLength || h == hTransferEncoding diff --git a/services/federator/test/unit/Test/Federator/Response.hs b/services/federator/test/unit/Test/Federator/Response.hs index 4a394c52083..1e1ff6da94e 100644 --- a/services/federator/test/unit/Test/Federator/Response.hs +++ b/services/federator/test/unit/Test/Federator/Response.hs @@ -15,34 +15,33 @@ -- You should have received a copy of the GNU Affero General Public License along -- with this program. If not, see . --- | Local reproduction harness for the "federation RPC comes back as a 200 with --- a non-JSON body" flake. +-- | Reproduction + regression harness for the "federation RPC comes back as a +-- 200 with a non-JSON body" flake. -- --- The federation @/rpc@ and @/federation@ endpoints are served over the same --- keep-alive HTTP/1.1 port as @/i/metrics@. The observed failure is a @POST --- /rpc/…@ returning a 200 whose body is an unrelated @/i/metrics@ page — i.e. a --- response the connection should never have produced for that request. That only --- happens if a response mis-frames the connection (leaving bytes behind) so a --- later request on the reused connection reads them. +-- The federator serves @/rpc@, @/federation@ and @/i/metrics@ over the same +-- keep-alive HTTP/1.1 port, and the integration suite hits them through a single +-- shared 'http-client' 'Manager' (connection pooling on). The observed failure +-- is a @POST /rpc/…@ returning a 200 whose body is an unrelated @/i/metrics@ +-- page. That can only happen if a response mis-frames the connection, so the +-- client reads past the response boundary into the next response on a reused +-- connection. -- --- Two angles are exercised here: +-- 'testForwardedContentLengthDesync' pins the actual cause: the outward HTTP/2 +-- response carries a @Content-Length@ header, and +-- 'Federator.Response.streamingResponseToWai' forwards it verbatim into a +-- 'Wai.responseStream'. Warp then HONOURS that length instead of chunking the +-- body it actually streams. If the declared length does not match the streamed +-- bytes (a truncated cold-start upstream, a reset mid-stream, a stale +-- @Content-Length@), the client reads the declared number of bytes and runs +-- straight into the following response. This test drives the REAL +-- 'streamingResponseToWai' and a byte-exact HTTP/1.1 client (which frames +-- exactly like 'http-client'): before the fix the RPC response is poisoned with +-- the next response's bytes; after the fix (strip framing headers, let Warp +-- frame the streamed body) the connection stays in sync. -- --- 1. 'testEmptyChunkDoesNotTruncate' drives the ACTUAL --- 'Federator.Response.streamingResponseToWai' with an upstream body that --- yields an empty chunk in the middle of the stream (which a streamed HTTP/2 --- upstream legitimately can). If Warp turned that into a premature chunked --- terminator, the response body would be truncated and the next pipelined --- request would desync. This pins down whether the streaming path can poison --- a keep-alive connection. --- --- 2. 'testConcurrentSharedManagerStaysInSync' stands up a federator-mimic (a --- chunked @/i/metrics@ and an @/rpc@ served through 'streamingResponseToWai') --- and hammers it through a SINGLE shared 'http-client' 'Manager' with a small --- connection pool — the same setup the integration suite uses — mixing metric --- scrapes and RPCs concurrently. It asserts no RPC response is ever poisoned --- by connection reuse. It is a control: if well-formed responses never --- desync here, the trigger must be a genuine mis-framing (see angle 1) rather --- than the shared Manager alone. +-- 'testEmptyChunkDoesNotTruncate' and 'testConcurrentSharedManagerStaysInSync' +-- are controls: well-framed streamed responses never desync, so the shared +-- Manager / connection pool is not itself the bug. module Test.Federator.Response (tests) where import Control.Concurrent.Async (forConcurrently) @@ -73,6 +72,9 @@ tests = testGroup "Response.streamingResponseToWai / keep-alive framing" [ testCase + "forwarding an upstream Content-Length must not desync the reused connection" + testForwardedContentLengthDesync, + testCase "empty upstream chunk does not truncate the streamed response or desync the connection" testEmptyChunkDoesNotTruncate, testCase @@ -96,36 +98,98 @@ jsonUpstream chunks = responseBody = source chunks } +-- | Like 'jsonUpstream', but also carrying a @Content-Length@ header — exactly +-- what the real outward HTTP/2 response carries. @declaredLen@ is what the +-- header claims; @chunks@ is what actually gets streamed. In production these +-- can diverge (truncated/cold-start upstream); here we set them apart on purpose +-- to model that. +jsonUpstreamWithContentLength :: Int -> [ByteString] -> StreamingResponse +jsonUpstreamWithContentLength declaredLen chunks = + Response + { responseStatusCode = HTTP.ok200, + responseHeaders = + Seq.fromList + [ (HTTP.hContentType, "application/json"), + (HTTP.hContentLength, BS8.pack (show declaredLen)) + ], + responseHttpVersion = HTTP.http20, + responseBody = source chunks + } + +-------------------------------------------------------------------------------- +-- The reproduction: a forwarded, mismatching Content-Length desyncs the +-- connection so the RPC request receives the *next* response (a metrics page). + +-- | A recognisable @/i/metrics@-style page, served the way the prometheus +-- middleware serves it (its own framing, on the same keep-alive port). +metricsPage :: ByteString +metricsPage = + "# HELP net_errors Number of exceptions caught by catchErrors middleware\n" + <> "# TYPE net_errors counter\n" + <> "net_errors 276.0\n" + +-- | @/first@ is the federation RPC (served via the real 'streamingResponseToWai' +-- from an upstream that over-declares its Content-Length); @/second@ is a metrics +-- page. These are the two things multiplexed over one keep-alive connection. +desyncApp :: Wai.Application +desyncApp req respond = + case Wai.rawPathInfo req of + "/first" -> + -- Upstream claims 100 bytes but only streams the (29-byte) body: a + -- truncated cold-start response. This is the RPC the probe issues. + respond (streamingResponseToWai (jsonUpstreamWithContentLength 100 [realFederationBody])) + _ -> + respond $ + Wai.responseBuilder + HTTP.ok200 + [(HTTP.hContentType, "text/plain; version=0.0.4")] + (byteString metricsPage) + +testForwardedContentLengthDesync :: Assertion +testForwardedContentLengthDesync = + Warp.testWithApplication (pure desyncApp) $ \port -> do + (r1, r2) <- pipelineTwo port + putStrLn $ + unlines + [ "", + "===== forwarded Content-Length desync =====", + "RPC (/first) response: " <> show r1, + "metrics (/second) response: " <> show r2 + ] + -- The RPC response must be exactly the JSON body the upstream streamed — + -- never contaminated with bytes from the following (metrics) response. + assertEqual "RPC response status" 200 (hrStatus r1) + assertEqual "RPC response content-type" (Just "application/json") (hrContentType r1) + assertBool + ( "RPC response body was poisoned by the next response on the reused connection: " + <> show (hrBody r1) + ) + (not ("# HELP" `BS.isInfixOf` hrBody r1) && not ("HTTP/1.1" `BS.isInfixOf` hrBody r1)) + assertEqual + "RPC response body must be exactly the streamed body (no over-read)" + realFederationBody + (hrBody r1) + -- ...and the following request's response must still be intact. + assertEqual "pipelined metrics response status" 200 (hrStatus r2) + assertEqual "pipelined metrics response content-type" (Just "text/plain; version=0.0.4") (hrContentType r2) + assertEqual "pipelined metrics response body" metricsPage (hrBody r2) + -------------------------------------------------------------------------------- --- Angle 1: empty chunk in the middle of a streamed body. +-- Control 1: empty chunk in the middle of a streamed body. --- | The upstream yields ["{\"a\":", "", "1}"] — an empty chunk between two --- non-empty ones. A conforming server must still deliver the concatenation --- ("{\"a\":1}") and keep the connection framed correctly. emptyChunkUpstream :: StreamingResponse emptyChunkUpstream = jsonUpstream ["{\"a\":", "", "1}"] testEmptyChunkDoesNotTruncate :: Assertion testEmptyChunkDoesNotTruncate = Warp.testWithApplication (pure (mkApp (streamingResponseToWai emptyChunkUpstream))) $ \port -> do - raw <- singleRequestRaw port "/first" (r1, r2) <- pipelineTwo port - putStrLn $ - unlines - [ "", - "===== empty-chunk framing =====", - "raw bytes of GET /first: " <> show raw, - "parsed 1st response: " <> show r1, - "parsed 2nd response: " <> show r2 - ] assertEqual "first response body must be the full concatenation of all chunks" "{\"a\":1}" (hrBody r1) assertEqual "second (pipelined) response must be intact (connection not desynced)" legitSecond r2 -------------------------------------------------------------------------------- --- Angle 2: shared Manager under concurrency (control). +-- Control 2: shared Manager under concurrency (well-framed responses). --- | A metrics page big enough to span several chunks, framed chunked (no --- Content-Length) just like the prometheus middleware serves @/i/metrics@. metricsBody :: ByteString metricsBody = "# HELP http_request_duration_seconds The HTTP request latencies in seconds.\n" @@ -149,8 +213,6 @@ federatorMimicApp req respond = testConcurrentSharedManagerStaysInSync :: Assertion testConcurrentSharedManagerStaysInSync = Warp.testWithApplication (pure federatorMimicApp) $ \port -> do - -- One shared Manager with a small pool, so RPC and metric requests are forced - -- to reuse the same handful of keep-alive connections (as in the suite). mgr <- HTTP.newManager HTTP.defaultManagerSettings {HTTP.managerConnCount = 4} let n = 400 :: Int base = "http://127.0.0.1:" <> show port @@ -170,14 +232,6 @@ testConcurrentSharedManagerStaysInSync = then Nothing else Just (i, status, ct, body) poisoned <- catMaybes <$> forConcurrently [1 .. n] oneRequest - putStrLn $ - "\n===== shared-Manager concurrency =====" - <> "\nRPC calls: " - <> show (length (filter odd [1 .. n])) - <> ", metric scrapes: " - <> show (length (filter even [1 .. n])) - <> ", poisoned RPC responses: " - <> show (length poisoned) for_ poisoned $ \p -> putStrLn (" POISONED: " <> show p) assertBool ("expected no RPC response to be poisoned by connection reuse, but got: " <> show poisoned) @@ -186,8 +240,6 @@ testConcurrentSharedManagerStaysInSync = -------------------------------------------------------------------------------- -- Test server plumbing. --- | Serve the given (already-rendered) response on @/first@; a distinct, plain --- JSON response on anything else (the "next request" whose integrity we check). mkApp :: Wai.Response -> Wai.Application mkApp firstResp req respond = case Wai.rawPathInfo req of @@ -209,8 +261,9 @@ legitSecond = -------------------------------------------------------------------------------- -- A minimal, conforming HTTP/1.1 client over a raw socket. It follows exactly --- the framing the server advertises (Content-Length, else chunked), which is --- where a mis-framed response bites a reused connection. +-- the framing the server advertises (Content-Length, else chunked) — the same +-- choice 'http-client' makes — which is where a mis-framed response bites a +-- reused connection. data HttpResponse = HttpResponse { hrStatus :: Int, @@ -220,7 +273,7 @@ data HttpResponse = HttpResponse deriving (Eq, Show) -- | Pipeline two requests on one keep-alive connection, then parse both --- responses in order (deterministic byte interleaving). +-- responses in order (deterministic byte interleaving — no timing races). pipelineTwo :: Int -> IO (HttpResponse, HttpResponse) pipelineTwo port = do addr <- resolve port @@ -233,21 +286,6 @@ pipelineTwo port = do r2 <- parseResponse buf pure (r1, r2) --- | Single request on a fresh connection; slurp bytes until the socket is idle --- (keep-alive means no EOF). Evidence only. -singleRequestRaw :: Int -> ByteString -> IO ByteString -singleRequestRaw port path = do - addr <- resolve port - bracket (open addr) close $ \sock -> do - sendAll sock $ "GET " <> path <> " HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n" - slurp sock mempty - where - slurp sock acc = do - mchunk <- timeout 400_000 (recv sock 4096) - case mchunk of - Just chunk | not (BS.null chunk) -> slurp sock (acc <> chunk) - _ -> pure acc - data Buf = Buf Socket (IORef ByteString) newBuf :: Socket -> IO Buf @@ -255,8 +293,10 @@ newBuf s = Buf s <$> newIORef mempty fill :: Buf -> IO Bool fill (Buf s ref) = do - more <- recv s 4096 - if BS.null more then pure False else True <$ modifyIORef' ref (<> more) + mmore <- timeout 1_000_000 (recv s 4096) + case mmore of + Just more | not (BS.null more) -> True <$ modifyIORef' ref (<> more) + _ -> pure False readLineB :: Buf -> IO ByteString readLineB buf@(Buf _ ref) = go @@ -332,10 +372,10 @@ readChunked buf acc = do sizeLine <- readLineB buf let sizeHex = BS8.takeWhile (/= ';') sizeLine case readHex (BS8.unpack sizeHex) of - [(0 :: Int, _)] -> acc <$ readLineB buf -- consume trailing CRLF after last chunk + [(0 :: Int, _)] -> acc <$ readLineB buf [(n, _)] -> do chunk <- readNB buf n - _ <- readNB buf 2 -- trailing CRLF after the chunk + _ <- readNB buf 2 readChunked buf (acc <> chunk) _ -> pure acc From a6001bce150feb1bbc19377ac5eaed7aa90c50a3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A9=20Schaul?= Date: Wed, 29 Jul 2026 09:30:58 +0000 Subject: [PATCH 08/10] federator: remove investigation diagnostic; regression test red pre-fix Drop the temporary callOutward diagnostic (it was on the wrong leg and generated false positives), and revert the streamingResponseToWai change so the new Test.Federator.Response regression test fails on CI, demonstrating the keep-alive response-framing desync before the fix lands. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../federator-streaming-response-framing | 1 - .../federator/src/Federator/InternalServer.hs | 19 --------------- services/federator/src/Federator/Response.hs | 23 +------------------ 3 files changed, 1 insertion(+), 42 deletions(-) delete mode 100644 changelog.d/3-bug-fixes/federator-streaming-response-framing diff --git a/changelog.d/3-bug-fixes/federator-streaming-response-framing b/changelog.d/3-bug-fixes/federator-streaming-response-framing deleted file mode 100644 index ae6fbdd737e..00000000000 --- a/changelog.d/3-bug-fixes/federator-streaming-response-framing +++ /dev/null @@ -1 +0,0 @@ -Federator: stop forwarding the upstream `Content-Length`/`Transfer-Encoding` headers when re-streaming a federation response. Warp re-frames the streamed body itself (chunked); forwarding a `Content-Length` made Warp serve the body under that declared length instead, so a truncated/reset cold-start upstream (declared length ≠ streamed bytes) desynchronised the caller's keep-alive connection and could deliver an unrelated response (e.g. a `/i/metrics` page in reply to a `POST /rpc/.../brig/api-version`). This was the cause of the flaky `Federation.testNotificationsForOfflineBackends` / `Federator.testFederatorNumRequestsMetrics` failures. diff --git a/services/federator/src/Federator/InternalServer.hs b/services/federator/src/Federator/InternalServer.hs index 2dfaee4e9c7..accd160987d 100644 --- a/services/federator/src/Federator/InternalServer.hs +++ b/services/federator/src/Federator/InternalServer.hs @@ -43,7 +43,6 @@ import Polysemy.TinyLog import Servant qualified import Servant.API import Servant.API.Extended.Endpath -import Servant.Client.Core (responseHeaders, responseStatusCode) import Servant.Server.Generic import System.Logger.Class qualified as Log import Wire.API.Federation.Component @@ -128,24 +127,6 @@ callOutward targetDomain component (RPC path) req cont = do path (Wai.requestHeaders req) (fromLazyByteString body) - -- Diagnostic for a known flaky failure mode (WPB: federation ingress probe - -- occasionally receives a 200 with a non-JSON body). The federator is a - -- transparent byte proxy, so a federation RPC coming back as a 200 whose body - -- is not JSON means the outward call reached something other than the target's - -- federation endpoint (e.g. an ingress/route not yet programmed during backend - -- startup, or a stale/mismatched response on a reused connection). We only - -- inspect the status line and Content-Type header here; the body must not be - -- consumed, as it is streamed straight back to the caller. - let respStatus = HTTP.statusCode (responseStatusCode resp) - respContentType = snd <$> find ((== HTTP.hContentType) . fst) (responseHeaders resp) - when (respStatus == 200 && respContentType /= Just "application/json") $ - warn $ - Log.msg (Log.val "Federator outward call returned a non-JSON 200; upstream is not the target's federation endpoint") - . Log.field "domain" targetDomain._domainText - . Log.field "component" (show component) - . Log.field "path" path - . Log.field "status" (show respStatus) - . Log.field "contentType" (fromMaybe ("" :: ByteString) respContentType) embed . cont $ streamingResponseToWai resp serveOutward :: Env -> Int -> IORef [IO ()] -> IO () diff --git a/services/federator/src/Federator/Response.hs b/services/federator/src/Federator/Response.hs index 4226d5dffef..7633252c287 100644 --- a/services/federator/src/Federator/Response.hs +++ b/services/federator/src/Federator/Response.hs @@ -19,32 +19,13 @@ module Federator.Response where import Data.ByteString.Builder import Imports -import Network.HTTP.Types.Header (hContentLength, hTransferEncoding) import Network.Wai qualified as Wai import Servant.Client.Core import Servant.Types.SourceT --- | Turn a streaming upstream response (from the outward federation call) into a --- WAI response that Warp serves back to the caller. --- --- We re-frame the body: Warp streams it with chunked transfer-encoding (it does --- not know the length up front). We must therefore DROP the upstream's own --- framing headers ('Content-Length', 'Transfer-Encoding'); forwarding them is a --- keep-alive desync waiting to happen. --- --- In particular, if we forward a 'Content-Length', Warp honours it verbatim and --- sends the streamed body raw under that declared length instead of chunking it. --- The moment the declared length disagrees with the number of bytes we actually --- stream — a truncated or reset cold-start upstream, a stale @Content-Length@ — --- the client reads exactly the declared number of bytes and runs straight past --- the response boundary into the next response on the reused connection. On the --- integration suite's shared, pooled HTTP/1.1 connection that surfaces as a --- @POST /rpc/…@ coming back with an unrelated @/i/metrics@ body. Stripping the --- framing headers lets Warp frame exactly what we stream, so the length on the --- wire can never disagree with the body and the connection stays in sync. streamingResponseToWai :: StreamingResponse -> Wai.Response streamingResponseToWai resp = - let headers = filter (not . isFramingHeader . fst) (toList (responseHeaders resp)) + let headers = toList (responseHeaders resp) status = responseStatusCode resp streamingBody output flush = foreach @@ -52,5 +33,3 @@ streamingResponseToWai resp = (\chunk -> output (byteString chunk) *> flush) (responseBody resp) in Wai.responseStream status headers streamingBody - where - isFramingHeader h = h == hContentLength || h == hTransferEncoding From 932315a0763644cad44eee62a3b1a0a821da22a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A9=20Schaul?= Date: Wed, 29 Jul 2026 09:48:53 +0000 Subject: [PATCH 09/10] federator: strip framing headers when re-streaming federation responses streamingResponseToWai forwarded the upstream response's Content-Length (and Transfer-Encoding) verbatim into a Wai.responseStream. Warp then honours that declared length instead of chunking the body it actually streams, so a truncated/reset cold-start upstream (declared length != streamed bytes) desynchronises the caller's keep-alive connection: the client reads the declared number of bytes and runs past the response boundary into the next response on the shared, pooled HTTP/1.1 connection. That surfaced as a POST /rpc/.../brig/api-version coming back 200 with an unrelated /i/metrics body, flaking Federation.testNotificationsForOfflineBackends and Federator.testFederatorNumRequestsMetrics. Drop the framing headers so Warp frames exactly what we stream; the length on the wire can never disagree with the body and the connection stays in sync. Turns the Test.Federator.Response regression test green. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../federator-streaming-response-framing | 1 + services/federator/src/Federator/Response.hs | 23 ++++++++++++++++++- 2 files changed, 23 insertions(+), 1 deletion(-) create mode 100644 changelog.d/3-bug-fixes/federator-streaming-response-framing diff --git a/changelog.d/3-bug-fixes/federator-streaming-response-framing b/changelog.d/3-bug-fixes/federator-streaming-response-framing new file mode 100644 index 00000000000..ae6fbdd737e --- /dev/null +++ b/changelog.d/3-bug-fixes/federator-streaming-response-framing @@ -0,0 +1 @@ +Federator: stop forwarding the upstream `Content-Length`/`Transfer-Encoding` headers when re-streaming a federation response. Warp re-frames the streamed body itself (chunked); forwarding a `Content-Length` made Warp serve the body under that declared length instead, so a truncated/reset cold-start upstream (declared length ≠ streamed bytes) desynchronised the caller's keep-alive connection and could deliver an unrelated response (e.g. a `/i/metrics` page in reply to a `POST /rpc/.../brig/api-version`). This was the cause of the flaky `Federation.testNotificationsForOfflineBackends` / `Federator.testFederatorNumRequestsMetrics` failures. diff --git a/services/federator/src/Federator/Response.hs b/services/federator/src/Federator/Response.hs index 7633252c287..4226d5dffef 100644 --- a/services/federator/src/Federator/Response.hs +++ b/services/federator/src/Federator/Response.hs @@ -19,13 +19,32 @@ module Federator.Response where import Data.ByteString.Builder import Imports +import Network.HTTP.Types.Header (hContentLength, hTransferEncoding) import Network.Wai qualified as Wai import Servant.Client.Core import Servant.Types.SourceT +-- | Turn a streaming upstream response (from the outward federation call) into a +-- WAI response that Warp serves back to the caller. +-- +-- We re-frame the body: Warp streams it with chunked transfer-encoding (it does +-- not know the length up front). We must therefore DROP the upstream's own +-- framing headers ('Content-Length', 'Transfer-Encoding'); forwarding them is a +-- keep-alive desync waiting to happen. +-- +-- In particular, if we forward a 'Content-Length', Warp honours it verbatim and +-- sends the streamed body raw under that declared length instead of chunking it. +-- The moment the declared length disagrees with the number of bytes we actually +-- stream — a truncated or reset cold-start upstream, a stale @Content-Length@ — +-- the client reads exactly the declared number of bytes and runs straight past +-- the response boundary into the next response on the reused connection. On the +-- integration suite's shared, pooled HTTP/1.1 connection that surfaces as a +-- @POST /rpc/…@ coming back with an unrelated @/i/metrics@ body. Stripping the +-- framing headers lets Warp frame exactly what we stream, so the length on the +-- wire can never disagree with the body and the connection stays in sync. streamingResponseToWai :: StreamingResponse -> Wai.Response streamingResponseToWai resp = - let headers = toList (responseHeaders resp) + let headers = filter (not . isFramingHeader . fst) (toList (responseHeaders resp)) status = responseStatusCode resp streamingBody output flush = foreach @@ -33,3 +52,5 @@ streamingResponseToWai resp = (\chunk -> output (byteString chunk) *> flush) (responseBody resp) in Wai.responseStream status headers streamingBody + where + isFramingHeader h = h == hContentLength || h == hTransferEncoding From fbb4ed847141d8e934bd4547fc86c6f730113f61 Mon Sep 17 00:00:00 2001 From: jschaul Date: Wed, 29 Jul 2026 12:03:17 +0200 Subject: [PATCH 10/10] Hi CI