Implement stateful per-chunk streaming compression - #2703
Closed
SavinduDimal wants to merge 5 commits into
Closed
Conversation
SavinduDimal
requested review from
AnuGayan,
Arshardh,
CrowleyRajapakse,
HeshanSudarshana,
HiranyaKavishani,
Krishanx92,
PasanT9,
RakhithaRR,
Tharsanan1,
Thushani-Jayasekera,
VirajSalaka,
ashera96,
chamilaadhi,
dushaniw,
hisanhunais,
lasanthaS,
malinthaprasan,
pubudu538,
renuka-fernando,
senthuran16,
tgtshanika,
tharikaGitHub,
tharindu1st and
thivindu
as code owners
July 15, 2026 12:00
Contributor
📝 WalkthroughWalkthroughChangesThe PR redesigns streaming decompression with coordinated shared state, adds persistent gzip/Brotli compression across chunks, integrates compressors into streaming request and response translation, and adds regression coverage. Streaming codec processing
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant StreamChunk
participant StreamingTranslator
participant PolicyExecutionContext
participant streamCompressor
participant CompressionWriter
StreamChunk->>StreamingTranslator: Translate streaming chunk
StreamingTranslator->>PolicyExecutionContext: Get or create compressor
StreamingTranslator->>streamCompressor: FeedChunk(outputBody, endOfStream)
streamCompressor->>CompressionWriter: Write and flush chunk
streamCompressor->>CompressionWriter: Close at end of stream
streamCompressor-->>StreamingTranslator: Return compressed output
StreamingTranslator-->>StreamChunk: Forward translated chunk
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@gateway/gateway-runtime/policy-engine/internal/kernel/translator.go`:
- Around line 1527-1533: The request recompression failure path in translator.go
around lines 1527-1533 must fail or terminate the stream instead of sending
uncompressed data; apply the same change to the response recompression failure
path around lines 1615-1621. Preserve the forwarded Content-Encoding state and
propagate the recompression error or terminate the corresponding
request/response stream rather than falling back to raw chunks.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 26680381-4b10-4551-99e1-c5776bd558a9
📒 Files selected for processing (4)
gateway/gateway-runtime/policy-engine/internal/kernel/decompression.gogateway/gateway-runtime/policy-engine/internal/kernel/decompression_test.gogateway/gateway-runtime/policy-engine/internal/kernel/execution_context.gogateway/gateway-runtime/policy-engine/internal/kernel/translator.go
SavinduDimal
force-pushed
the
stream-compressor
branch
from
July 22, 2026 13:53
d33ad8b to
711023c
Compare
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@gateway/gateway-runtime/policy-engine/internal/kernel/decompression.go`:
- Around line 67-78: Bound decompressed output in streamDecompressor instead of
retaining an unbounded bytes.Buffer. Update the decoder/output flow used by
FeedChunk to enforce maxStreamAccumulatorSize while preserving chunked
consumption and terminal error behavior, and avoid copying decoded payloads
beyond the configured cap.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 6dbe6d62-8a49-4c01-b1b3-eb6cd0e247cc
📒 Files selected for processing (4)
gateway/gateway-runtime/policy-engine/internal/kernel/decompression.gogateway/gateway-runtime/policy-engine/internal/kernel/decompression_test.gogateway/gateway-runtime/policy-engine/internal/kernel/execution_context.gogateway/gateway-runtime/policy-engine/internal/kernel/translator.go
🚧 Files skipped from review as they are similar to previous changes (1)
- gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go
Implement streamCompressor for stateful, per-chunk streaming compression
SavinduDimal
force-pushed
the
stream-compressor
branch
from
July 24, 2026 13:37
7e8c693 to
47a7d90
Compare
Contributor
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go (1)
858-875: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winClose the response decompressor on error paths.
The new cleanup runs only after
ExecuteStreamingResponsePoliciessucceeds. If it returns an error at Line 859, the function exits beforeresponseStreamDecomp.Close(), leaving the decoder goroutine blocked. A translation error from Line 875 can cause the same leak. Centralize cleanup and invoke it before every terminal/error return.Proposed cleanup flow
+ closeResponseDecomp := func() { + if ec.responseStreamDecomp != nil { + ec.responseStreamDecomp.Close() + ec.responseStreamDecomp = nil + } + } + execResult, err := ec.server.executor.ExecuteStreamingResponsePolicies( ctx, ec.policyChain.Policies, @@ ) if err != nil { + closeResponseDecomp() return ec.handlePolicyError(ctx, err, "response_body_streaming"), nil } @@ - return TranslateStreamingResponseChunkAction(execResult, chunk, ec) + resp, err := TranslateStreamingResponseChunkAction(execResult, chunk, ec) + if err != nil { + closeResponseDecomp() + } + return resp, err🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go` around lines 858 - 875, Centralize response decompressor cleanup in the streaming response execution flow, using the existing responseStreamDecomp lifecycle and terminal conditions. Ensure cleanup runs before the handlePolicyError return when ExecuteStreamingResponsePolicies fails, before returning any TranslateStreamingResponseChunkAction error, and on normal EndOfStream or StreamTerminated completion; keep it idempotent and clear the field after closing.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@gateway/gateway-runtime/policy-engine/internal/kernel/execution_context.go`:
- Around line 858-875: Centralize response decompressor cleanup in the streaming
response execution flow, using the existing responseStreamDecomp lifecycle and
terminal conditions. Ensure cleanup runs before the handlePolicyError return
when ExecuteStreamingResponsePolicies fails, before returning any
TranslateStreamingResponseChunkAction error, and on normal EndOfStream or
StreamTerminated completion; keep it idempotent and clear the field after
closing.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 5c5b67d6-e25d-488e-bb25-1d9d833b6f9d
📒 Files selected for processing (4)
gateway/gateway-runtime/policy-engine/internal/kernel/decompression.gogateway/gateway-runtime/policy-engine/internal/kernel/decompression_test.gogateway/gateway-runtime/policy-engine/internal/kernel/execution_context.gogateway/gateway-runtime/policy-engine/internal/kernel/translator.go
🚧 Files skipped from review as they are similar to previous changes (3)
- gateway/gateway-runtime/policy-engine/internal/kernel/translator.go
- gateway/gateway-runtime/policy-engine/internal/kernel/decompression_test.go
- gateway/gateway-runtime/policy-engine/internal/kernel/decompression.go
Closed
2 tasks
Contributor
|
Related PR: #2878 |
Contributor
Author
|
Closing this PR since the issues are not reproducible after PR: #2878 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
Issue 1: Analytics + Streaming
streamCompressorkeeps one writer alive for the life of the stream, flushing (Z_SYNC_FLUSH) after each chunk and closing only at EndOfStream, producing a single continuous member the client decodes in full.Issue 2: Timeout error
Then found a new issue when we work with large responses, response flow got stalled and faced timeout errors.
The old streamDecompressor used an io.Pipe + decoder goroutine + a bounded output channel (cap 64). On long/large responses the channel filled, the decoder blocked on the send, stopped reading the pipe, and the next FeedChunk blocked forever on pw.Write — a back-pressure deadlock. This is the mid-stream stall (stream would forward ~part of the body, then go silent until the client timed out). It only showed up on large tasks, which is why short prompts worked and the original 2-chunk tests missed it.
A minimal patch could stop the deadlock, but the io.Pipe design has a deeper flaw: pw.Write returning tells you bytes were copied to the reader, not decoded, so FeedChunk had no reliable signal for "all output for this chunk is ready" — it relied on a racy runtime.Gosched() + best-effort drain. Getting a correct signal on top of the pipe means reimplementing exactly what the refactor introduces, so the pipe becomes dead weight.
The refactor replaces io.Pipe + channel + errChan + Gosched with a single sync.Cond + a feederReader that records when the decoder has consumed all fed input and parked. FeedChunk now returns exactly when every decodable byte is available — deadlock-free and deterministic — and ends up with fewer moving parts than before.