Repository navigation
Replace the read cache with pure request coalescing - #121
Merged
Merged
Conversation
The read cache stored response bodies for 60s (10min for the catalog) and replayed them. Its correctness rested on a cache key covering every dimension a response depends on, an open-ended invariant that has already been broken twice: a path-only key served page 1 of the plugin catalog to page-2 requests (codex looped on its own cursor: 3,598 requests, 15 GB RSS, kernel panics) and served one account's installed plugins to another. The loop guard, TTL, eviction, and entry cap all existed to fence that stored state. Delete the stored state and keep only singleflight: identical concurrent requests to the plugin polling endpoints share one upstream fetch, nothing is retained afterward, so there is no staleness window and nothing to guard. Non-concurrent polls pass through, which is traffic parity with no proxy at all. The full-request flightKey and its key-completeness property test carry over unchanged. Catalog aggregation stays (codex OOMs by pages walked and its disk cache is cold for any proxy base URL), but a walk that fails midway or hits the page/byte cap now returns an error instead of the pages collected so far. A partial merged catalog carries no continuation token, so clients cache it as complete; codex pinned 173- and 209-entry catalogs (of 2,285) for 3 hours after upstream throttled a walk. A 502 gets retried; wrong data does not. Net -271 lines.
📝 WalkthroughWalkthroughThe proxy replaces persistent GET response caching with request coalescing for selected plugin polling paths. It adds caller-aware flight keys, buffers responses for concurrent waiters, and returns errors for failed or partial catalog aggregation. ChangesProxy request coalescing
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant CacheFlight
participant aggregateCatalogPages
participant UpstreamProxy
Client->>CacheFlight: Submit identical polling request
CacheFlight->>UpstreamProxy: Fetch upstream response
UpstreamProxy-->>aggregateCatalogPages: Return catalog pages
aggregateCatalogPages-->>CacheFlight: Return aggregated response or error
CacheFlight-->>Client: Replay status, headers, and body
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 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}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 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 `@internal/proxy/catalog_aggregate.go`:
- Around line 146-152: The aggregation function should propagate errors from
every JSON unmarshal or marshal failure instead of returning nil errors with
zeroed page and plugin counts. Update the failure paths around json.Unmarshal,
json.Marshal, and the later marshal at the function’s final response
construction to return the encountered error while preserving the existing
firstBody, pages, and merged-count values required by the caller’s failed-walk
policy.
In `@internal/proxy/proxy.go`:
- Around line 2056-2091: Update the CacheFlight.do flight work around
aggregateCatalogPages to use a request derived with
context.WithoutCancel(proxyRequest.Context()), so the shared catalog walk is not
canceled when the initiating client disconnects. Apply the transport and
cancellation timeouts to this detached request context, then pass that request
through rp.ServeHTTP and aggregateCatalogPages while preserving the existing
response and error handling.
In `@internal/proxy/request_coalesce.go`:
- Around line 40-54: Update flightKey to include the request’s Accept-Encoding
value as an additional key dimension, preserving distinct coalescing flights for
callers with different negotiated encodings. Add the corresponding coverage to
TestFlightKeyCoversEveryRequestDimension.
🪄 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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 3ea833c9-1111-4d61-bb75-1cab921a024d
📒 Files selected for processing (10)
internal/proxy/catalog_aggregate.gointernal/proxy/catalog_aggregate_test.gointernal/proxy/multitenant.gointernal/proxy/proxy.gointernal/proxy/read_cache.gointernal/proxy/read_cache_regression_test.gointernal/proxy/read_cache_test.gointernal/proxy/request_coalesce.gointernal/proxy/request_coalesce_test.gointernal/proxy/session_lease_test.go
💤 Files with no reviewable changes (4)
- internal/proxy/read_cache.go
- internal/proxy/session_lease_test.go
- internal/proxy/read_cache_test.go
- internal/proxy/read_cache_regression_test.go
Comment on lines
146
to
+152
| var payload map[string]json.RawMessage | ||
| if err := json.Unmarshal(firstBody, &payload); err != nil { | ||
| return firstBody, 0, 0, false | ||
| return firstBody, 0, 0, false, nil | ||
| } | ||
| mergedPlugins, err := json.Marshal(merged) | ||
| if err != nil { | ||
| return firstBody, 0, 0, false | ||
| return firstBody, 0, 0, false, nil |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Return an error from the marshalling failure paths.
Lines 147, 150, and 164 detect an error and then return nil. golangci-lint reports both sites under nilerr, which fails the lint gate. The values are also inconsistent with the walk that already ran: pages and len(merged) are discarded and reported as 0.
The returned firstBody still carries its next_page_token, so the client is not handed a partial catalog dressed as complete. The behaviour is therefore safe, but it contradicts the rule stated in the doc comment on lines 69-73. Return the error so the caller applies one policy for every failed walk.
♻️ Proposed fix for the swallowed marshalling errors
var payload map[string]json.RawMessage
if err := json.Unmarshal(firstBody, &payload); err != nil {
- return firstBody, 0, 0, false, nil
+ return nil, pages, len(merged), false, fmt.Errorf("catalog first page: %w", err)
}
mergedPlugins, err := json.Marshal(merged)
if err != nil {
- return firstBody, 0, 0, false, nil
+ return nil, pages, len(merged), false, fmt.Errorf("marshal merged plugins: %w", err)
} out, err := json.Marshal(payload)
if err != nil {
- return firstBody, 0, 0, false, nil
+ return nil, pages, len(merged), false, fmt.Errorf("marshal merged catalog: %w", err)
}Note: TestAggregateIgnoresUnrelatedResponses covers the "not json" case through the early-return path on lines 90-92, so it still passes.
Also applies to: 164-168
🧰 Tools
🪛 golangci-lint (2.12.2)
[error] 148-148: error is not nil (line 147) but it returns nil
(nilerr)
[error] 152-152: error is not nil (line 150) but it returns nil
(nilerr)
🤖 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 `@internal/proxy/catalog_aggregate.go` around lines 146 - 152, The aggregation
function should propagate errors from every JSON unmarshal or marshal failure
instead of returning nil errors with zeroed page and plugin counts. Update the
failure paths around json.Unmarshal, json.Marshal, and the later marshal at the
function’s final response construction to return the encountered error while
preserving the existing firstBody, pages, and merged-count values required by
the caller’s failed-walk policy.
Source: Linters/SAST tools
Comment on lines
+2056
to
+2091
| flight, _ := s.CacheFlight.do(flightKey(r), func() flightResult { | ||
| rec := &responseRecorder{} | ||
| rp.ServeHTTP(rec, proxyRequest) | ||
| body := rec.buf.Bytes() | ||
| if rec.code >= 200 && rec.code < 300 { | ||
| merged, pages, entries, ok, err := aggregateCatalogPages(transport, proxyRequest, upstream, body) | ||
| if err != nil { | ||
| // A partial catalog has no continuation token, so the | ||
| // client would cache it as complete (codex pins it on | ||
| // disk for 3 hours). Fail the request instead; a retry | ||
| // is cheap, undoing wrong data is not. | ||
| if s.Logger != nil { | ||
| s.Logger.Warn("catalog aggregation failed", "account", account.ID, | ||
| "path", r.URL.Path, "pages", pages, "entries", entries, "error", err) | ||
| } | ||
| return flightResult{ | ||
| statusCode: http.StatusBadGateway, | ||
| header: http.Header{"Content-Type": []string{"text/plain; charset=utf-8"}}, | ||
| body: []byte("catalog aggregation failed upstream\n"), | ||
| } | ||
| } | ||
| header := rec.header | ||
| if header == nil { | ||
| header = make(http.Header) | ||
| if ok { | ||
| body = merged | ||
| if s.Logger != nil { | ||
| s.Logger.Info("aggregated plugin catalog pages", "account", account.ID, | ||
| "path", r.URL.Path, "pages", pages, "entries", entries) | ||
| } | ||
| } | ||
| header.Del("Content-Length") | ||
| return flightResult{statusCode: rec.code, header: header, body: body} | ||
| }) | ||
| rec := &cacheRecorder{code: flight.statusCode, header: flight.header} | ||
| if rec.header == nil { | ||
| rec.header = make(http.Header) | ||
| } | ||
| rec.buf.Write(flight.body) | ||
| payload := rec.buf.Bytes() | ||
| if rec.code >= 200 && rec.code < 300 && !shared { | ||
| s.ReadCache.set(r, rec.code, rec.header, payload, cacheTTL) | ||
| } | ||
| for k, vs := range rec.header { | ||
| for _, v := range vs { | ||
| w.Header().Add(k, v) | ||
| } | ||
| header := rec.header | ||
| if header == nil { | ||
| header = make(http.Header) | ||
| } | ||
| if rec.code != 0 { | ||
| w.WriteHeader(rec.code) | ||
| header.Del("Content-Length") | ||
| return flightResult{statusCode: rec.code, header: header, body: body} | ||
| }) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Inspect singleFlight.do to learn what its second return value means.
rg -n -B5 -A45 'func \(.*singleFlight\) do' --type=goRepository: manaflow-ai/subrouter
Length of output: 159
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "tracked files matching proxy.go:"
git ls-files | rg '(^|/)proxy\.go$' || true
echo
echo "context and singleFlight occurrences:"
rg -n 'singleFlight|CacheFlight|do\(|WithoutCancel|RequestWithContext|ServeHTTP\('\
--type=go . || true
echo
echo "if internal/proxy/proxy.go exists, outline and relevant lines:"
if [ -f internal/proxy/proxy.go ]; then
wc -l internal/proxy/proxy.go
ast-grep outline internal/proxy/proxy.go --match singleFlight --view expanded || true
sed -n '1920,2125p' internal/proxy/proxy.go
fiRepository: manaflow-ai/subrouter
Length of output: 19144
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "tracked files matching proxy.go:"
git ls-files | rg '(^|/)proxy\.go$' || true
echo
echo "context and singleFlight occurrences:"
rg -n 'singleFlight|CacheFlight|do\(|WithoutCancel|RequestWithContext|ServeHTTP\('\
--type=go . || true
echo
echo "if internal/proxy/proxy.go exists, outline and relevant lines:"
if [ -f internal/proxy/proxy.go ]; then
wc -l internal/proxy/proxy.go
ast-grep outline internal/proxy/proxy.go --match singleFlight --viewA expanded || true
sed -n '1920,2125p' internal/proxy/proxy.go
fiRepository: manaflow-ai/subrouter
Length of output: 19296
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "singleFlight implementation:"
sed -n '1,90p' internal/proxy/single_flight.go
echo
echo "singleFlight tests around result/shared handling:"
sed -n '1,115p' internal/proxy/single_flight_test.go
echo
echo "coalescablePath implementations:"
rg -n -B5 -A20 'func coalescablePath|coalescablePath' --type=go .
echo
echo "proxyRequest creation context area:"
sed -n '800,945p' internal/proxy/proxy.go
echo
echo "aggregation call context:"
rg -n -B5 -A25 'func aggregateCatalogPages|aggregateCatalogPages\(' --type=go internalRepository: manaflow-ai/subrouter
Length of output: 29379
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "proxyRequest construction around handler start:"
rg -n -B80 -A80 'proxyRequest,.*context|proxyRequest, http.NoBody|ServeHTTP\(' internal/proxy/proxy.go | sed -n '1,220p'
echo
echo "catalog_aggregate loop body:"
sed -n '69,150p' internal/proxy/catalog_aggregate.goRepository: manaflow-ai/subrouter
Length of output: 11216
Detach the catalog coalescing flight from the initiating request context.
singleFlight.do runs the same upstream page walk for every caller at the same key. Calling it with proxyRequest.Context() makes the leader’s HTTP request cancellable by that caller, and aggregateCatalogPages clones that request context before each RoundTrip. One client that disconnects can cancel the shared catalog walk and return a cached flightResult for the remaining callers.
Use context.WithoutCancel(proxyRequest.Context()) for the flight work and put transport/cancellation timeouts under control in that request.
🤖 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 `@internal/proxy/proxy.go` around lines 2056 - 2091, Update the CacheFlight.do
flight work around aggregateCatalogPages to use a request derived with
context.WithoutCancel(proxyRequest.Context()), so the shared catalog walk is not
canceled when the initiating client disconnects. Apply the transport and
cancellation timeouts to this detached request context, then pass that request
through rp.ServeHTTP and aggregateCatalogPages while preserving the existing
response and error handling.
Comment on lines
+40
to
+54
| // flightKey identifies everything a shared response depends on: the method, | ||
| // the path, every query parameter, and the calling identity. Anything left out | ||
| // here is a request that gets served somebody else's body. | ||
| func flightKey(r *http.Request) string { | ||
| var b strings.Builder | ||
| b.WriteString(r.Method) | ||
| b.WriteByte('\n') | ||
| b.WriteString(r.URL.Path) | ||
| b.WriteByte('\n') | ||
| // Encode() sorts by key, so parameter order does not fragment the key space. | ||
| b.WriteString(r.URL.Query().Encode()) | ||
| b.WriteByte('\n') | ||
| b.WriteString(callerIdentity(r)) | ||
| return b.String() | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
flightKey omits content-negotiation headers, so a waiter can receive an unacceptable body encoding.
The key covers method, path, query, and caller identity. It does not cover Accept-Encoding. httputil.ReverseProxy forwards the client's Accept-Encoding upstream, so the buffered body carries whatever encoding the first caller negotiated, together with the origin's Content-Encoding header.
Two concurrent callers of the same endpoint with the same identity can differ here. Codex sends Accept-Encoding: gzip; a plain curl or a client without gzip support does not. The second caller then joins the flight and receives a gzip body it declared it cannot decode. proxy.go lines 2092-2096 copy every response header, including Content-Encoding, so the mismatch reaches the client.
The same gap affects catalog aggregation: aggregateCatalogPages parses the buffered bytes as JSON. If the body is gzip-encoded, catalogPlugins fails on page 1 and aggregation silently does not run.
Add the negotiated encoding to the key.
🐛 Proposed fix to include the encoding dimension
func flightKey(r *http.Request) string {
var b strings.Builder
b.WriteString(r.Method)
b.WriteByte('\n')
b.WriteString(r.URL.Path)
b.WriteByte('\n')
// Encode() sorts by key, so parameter order does not fragment the key space.
b.WriteString(r.URL.Query().Encode())
b.WriteByte('\n')
+ // A shared body carries one Content-Encoding. Callers that negotiated a
+ // different encoding must not join the same flight.
+ b.WriteString(r.Header.Get("Accept-Encoding"))
+ b.WriteByte('\n')
b.WriteString(callerIdentity(r))
return b.String()
}An alternative is to strip Accept-Encoding from proxyRequest before the flight and to request an identity encoding, which also lets aggregateCatalogPages parse the body reliably. Add a case to TestFlightKeyCoversEveryRequestDimension for whichever option you pick.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // flightKey identifies everything a shared response depends on: the method, | |
| // the path, every query parameter, and the calling identity. Anything left out | |
| // here is a request that gets served somebody else's body. | |
| func flightKey(r *http.Request) string { | |
| var b strings.Builder | |
| b.WriteString(r.Method) | |
| b.WriteByte('\n') | |
| b.WriteString(r.URL.Path) | |
| b.WriteByte('\n') | |
| // Encode() sorts by key, so parameter order does not fragment the key space. | |
| b.WriteString(r.URL.Query().Encode()) | |
| b.WriteByte('\n') | |
| b.WriteString(callerIdentity(r)) | |
| return b.String() | |
| } | |
| // flightKey identifies everything a shared response depends on: the method, | |
| // the path, every query parameter, and the calling identity. Anything left out | |
| // here is a request that gets served somebody else's body. | |
| func flightKey(r *http.Request) string { | |
| var b strings.Builder | |
| b.WriteString(r.Method) | |
| b.WriteByte('\n') | |
| b.WriteString(r.URL.Path) | |
| b.WriteByte('\n') | |
| // Encode() sorts by key, so parameter order does not fragment the key space. | |
| b.WriteString(r.URL.Query().Encode()) | |
| b.WriteByte('\n') | |
| // A shared body carries one Content-Encoding. Callers that negotiated a | |
| // different encoding must not join the same flight. | |
| b.WriteString(r.Header.Get("Accept-Encoding")) | |
| b.WriteByte('\n') | |
| b.WriteString(callerIdentity(r)) | |
| return b.String() | |
| } |
🤖 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 `@internal/proxy/request_coalesce.go` around lines 40 - 54, Update flightKey to
include the request’s Accept-Encoding value as an additional key dimension,
preserving distinct coalescing flights for callers with different negotiated
encodings. Add the corresponding coverage to
TestFlightKeyCoversEveryRequestDimension.
E2E against the live catalog: codex requests limit=50, and 2,283 entries is 46 pages, over the old cap of 40. With truncation now an error, the cap being under the real catalog size turned every cold catalog fetch into a 502. 256 pages leaves headroom while the 128 MB byte bound still terminates a cursor-rotating upstream.
The incident lived in the gap between correct units: the read cache passed its own tests while the handler replayed page 1 of the catalog to page-2 requests. These tests drive the real handler end to end: - page-2 with page-1's cursor gets page 2, never page 1 (the incident) - sequential identical requests each reach upstream, and X-Subrouter-Cache never reappears (the no-stored-responses contract) - N concurrent catalog requests cost one upstream walk and every waiter gets the complete merged catalog (both halves of coalescing) - a mid-walk failure is a 5xx, never a 2xx partial body (codex pins any 2xx in a 3-hour disk cache) - inference paths stay out of the buffering path (SSE would break) Mutation-verified: removing the query-string dimension from flightKey fails TestFlightKeyCoversEveryRequestDimension, TestPluginListPageTwoIsNeverPageOne, and TestConcurrentCatalogRequestsShareOneCompleteWalk.
lawrencecchen
added a commit
that referenced
this pull request
Jul 31, 2026
Three findings from the #121 review, each with a pinned test: - flightKey now includes Accept-Encoding. The reverse proxy forwards it upstream, so the buffered body carries whatever Content-Encoding the flight's leader negotiated; a waiter that never offered gzip would have received gzip bytes. Added as a dimension to the key-completeness test. - The shared walk runs on context.WithoutCancel of the leader's request. One disconnecting client used to cancel the flight for every waiter. TestLeaderDisconnectDoesNotCancelSharedWalk cancels the leader mid-walk and asserts the walk still completes. - aggregateCatalogPages returns errors from its marshalling failure paths instead of nil (nilerr), so every failed walk takes the same 502 path.
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
1 participant
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.
The read cache stored response bodies and replayed them for a TTL. Its correctness rested on the cache key covering every dimension a response depends on, an invariant already broken twice: the path-only key served page 1 of the plugin catalog to page-2 requests (#105: 3,598 requests in one turn, 5.3 GB streamed, 15 GB client RSS, kernel panics) and served one account's plugins to another. The loop guard, TTL, eviction, and entry cap were all fences around that stored state.
This deletes the stored state and keeps only singleflight: identical concurrent requests to the plugin polling endpoints share one upstream fetch and nothing is retained afterward. No staleness window, nothing to guard. Non-concurrent polls pass through, which is traffic parity with the no-proxy world. The full-request
flightKeyand the key-completeness property test from #105 carry over unchanged.Catalog aggregation (#107) stays, since codex OOMs by pages walked and its disk cache is cold for any proxy base URL. One behavior change: a walk that fails midway or exceeds the page/byte cap now returns 502 instead of the pages collected so far. A partial merged catalog has no continuation token, so clients cache it as complete; codex pinned 173- and 209-entry catalogs (of 2,285) on disk for 3 hours after upstream throttled a walk. A failed request gets retried; plausible wrong data does not.
Removed:
read_cache.go, its tests, theReadCacheserver field, theX-Subrouter-Cacheheader. Net -271 lines.go build,go vet,go test ./...all pass.Need help on this PR? Tag
@codesmith-botwith what you need. Autofix is disabled.Summary by cubic
Replaces the stored read cache with pure request coalescing for plugin polling endpoints to eliminate stale/wrong data while keeping burst protection. Adds end-to-end tests to pin plugin endpoint behavior, and raises the catalog aggregation page cap to 256 to fit the live catalog.
Refactors
Server.ReadCache, and theX-Subrouter-Cacheheader.Server.CacheFlight,coalescablePath,flightKey, and aresponseRecorder; only GET plugin polling paths are coalesced. Catalog aggregation runs inside the flight and stripsContent-Length.Bug Fixes
flightKeycovers method, path, query, and caller identity; handler-level regression tests ensure page-2 never replays page-1, no stored responses between requests, concurrent clients share one complete catalog walk, and inference paths are not coalesced.Written for commit 934f1dd. Summary will update on new commits.
Summary by CodeRabbit
New Features
Bug Fixes
502 Bad Gateway.Reliability