Fix stale SSE heights spawning concurrent /height polls#1271
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
📝 WalkthroughWalkthroughAdds EnvioApiClient.make to inject a hyperindex/ User-Agent into Rest clients, switches HyperFuel/HyperSync to use it, refactors SourceManager's stale-subscription fallback to a jittered Promise.race between subscription wait and a single fallback poll loop, and adds a regression test preventing concurrent fallback polls. ChangesStale subscription fallback polling fix
Sequence Diagram(s)sequenceDiagram
participant SSE as SSESubscription
participant SM as SourceManager
participant PHR as PendingHeightResolvers
participant SRC as Source.getHeightOrThrow
SSE->>SM: emit height (may be stale)
SM->>PHR: wait for pending resolver -> height > knownHeight
SM->>SRC: start fallback poll after stallTimeout/2 + jitter
SRC-->>SM: return height or error
PHR-->>SM: resolved height (> knownHeight)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related issues
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. Comment |
…heights (#1270) Add a test reproducing the per-process /height request amplification: when the SSE height stream re-emits non-increasing heights (e.g. a flapping/reconnecting stream re-sending the current head), each event spins the outer wait loop in getSourceNewHeight and leaks another uncancelled pollingFallback. The leaked loops accumulate and each polls at pollingInterval, so N stale events produce ~N concurrent /height poll loops instead of one bounded loop. The test drives 20 stale events and observes 21 concurrent polls, demonstrating the effective request rate scales with stale-event count and is not bounded by pollingInterval. https://claude.ai/code/session_01MCsYorVnvp3AKFtbyaiefw
…poll (#1270) The SSE->REST height fallback created a fresh pollingFallback promise on every outer-loop iteration. When the height stream re-emitted non-increasing heights (e.g. a reconnecting stream re-sending the current head, which onHeight does not filter), the loop spun and leaked an uncancelled poller each time. The leaked loops accumulated and each polled at pollingInterval (100ms), so a single stalled chain could drive thousands of /height req/s from one process and starve the shared HTTP client for every other chain. - Replace the per-iteration fallback with one subscription waiter and one REST poller, each looping internally until it sees a height > knownHeight. A burst of stale events can no longer spawn additional pollers. - Poll the fallback at stalledPollingInterval with jitter instead of the raw 100ms pollingInterval, so the no-subscription and subscription-stall paths behave consistently and many indexers don't poll in lockstep. - Both waiters stop when status becomes Done, so an orphaned poller doesn't keep polling after another source wins. Also tag the REST /height poll with the hyperindex User-Agent (HyperSync and HyperFuel) via a shared EnvioApiClient, matching the SSE stream and Rust data client so the requests are attributable instead of the default node agent. https://claude.ai/code/session_01MCsYorVnvp3AKFtbyaiefw
JonoPrest
force-pushed
the
claude/gifted-newton-NH7Yn
branch
from
1c341fa to
874333d
Compare
| sourceState.pendingHeightResolvers->Array.push(resolve) | ||
| }) | ||
| // If subscription goes quiet for half the stall timeout, fall back to REST polling | ||
| let pollingFallback = Utils.delay(stallTimeout / 2)->Promise.then(async () => { |
There was a problem hiding this comment.
I think you should only update Utils.delay(stallTimeout / 2) here, so it runs in a random order. Other changes in the files look irrelevant. status.done is not needed, since h.contents <= knownHeight is enough.
There was a problem hiding this comment.
This isn't a lockstep timing issue, it's a leak. The outer while creates a new
pollingFallback on every iteration. A stale height resolves the race through the
subscription but doesn't advance newHeight, so the loop runs again and creates
another pollingFallback, and the previous one is never cancelled. So N stale events
give N concurrent poll loops, each hitting /height every pollingInterval. That's
the runaway we saw, around 1.2 to 2.9k req/s from one process rather than 10.
Jittering stallTimeout / 2 only randomizes when each leaked poller starts, not how
many there are. 20 stale events still spawn 20 pollers. The regression test shows
this. 20 stale events give 21 concurrent /height calls on the current code and 1
with the single poller. Jitter alone still fails it.
The stale events are real. /height/sse is a WatchStream, so every reconnect
re-emits the current head, which is at or below what we already know, more so when LB
instances drift.
So the restructure is the fix, not the jitter. Trigger jitter is still worth having
on top, so I added it.
On status !== Done, agreed for the subscription waiter since it's parked on the
await, so I dropped it there. I kept it in the poll loop to stop an orphaned poller
hammering a drifted instance forever after another source has won. h <= knownHeight
doesn't cover that case. Added a comment. Can drop it if you'd rather.
UA/EnvioApiClient is defect 1 from the issue. Can split it into its own PR if you
want this one scoped to the polling fix.
There was a problem hiding this comment.
Do you mean that the actual issue that pendingHeightResolvers resolves with the current height, not increased one?
There was a problem hiding this comment.
Yes exactly. Do you think we should just filter?
There was a problem hiding this comment.
I think we need to fix the line instead - by adding a protective if that newHeight is not lower than sourceState.knownHeight
let unsubscribe = createSubscription(~onHeight=newHeight => {
sourceState.knownHeight = newHeight
There was a problem hiding this comment.
Much cleaner, will do 👍🏼
…eck in subscription waiter - Jitter the fallback poll trigger across [stallTimeout/2, stallTimeout) so indexers that go quiet together don't start polling in lockstep. - Drop the status check from the subscription waiter (it's parked on the await there, so the guard never runs). Keep it in the poll loop, where it stops an orphaned poller from hammering a lagging instance after another source wins. - Widen the two timing-sensitive fallback tests to wait past the jittered trigger. https://claude.ai/code/session_01MCsYorVnvp3AKFtbyaiefw
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
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)
packages/envio/src/sources/SourceManager.res (1)
346-383:⚠️ Potential issue | 🟠 Major | ⚡ Quick winRemove the raced subscription resolver when REST wins.
The resolver pushed on Line 352 is never removed if
pollFallback()wins the race on Line 383. In the exact “stale/quiet SSE, REST keeps finding blocks” path this PR targets, eachwaitForNewBlockcall leaves one more abandoned resolver inpendingHeightResolvers, so the array can grow unbounded until some future SSE event happens to flush it.🧹 Suggested cleanup
- let waitForSubscription = async () => { - let h = ref(initialHeight) - while h.contents <= knownHeight { - h := - ( - await Promise.make((resolve, _reject) => { - sourceState.pendingHeightResolvers->Array.push(resolve) - }) - ) - } - h.contents - } + let waitForSubscription = () => { + let resolverRef = ref(None) + let wait = async () => { + let h = ref(initialHeight) + while h.contents <= knownHeight { + h := await Promise.make((resolve, _reject) => { + resolverRef := Some(resolve) + sourceState.pendingHeightResolvers->Array.push(resolve) + }) + } + h.contents + } + let cleanup = () => + switch resolverRef.contents { + | Some(resolve) => + sourceState.pendingHeightResolvers = + sourceState.pendingHeightResolvers->Array.keep(r => r !== resolve) + | None => () + } + (wait(), cleanup) + } let pollFallback = async () => { let half = stallTimeout / 2 await Utils.delay(half + (Math.random() *. half->Belt.Int.toFloat)->Belt.Float.toInt) logger->Logging.childTrace({ "msg": "onHeight subscription stale, switching to polling fallback", @@ - let height = await Promise.race([waitForSubscription(), pollFallback()]) + let (subscriptionPromise, cleanupSubscriptionWait) = waitForSubscription() + let height = await Promise.race([subscriptionPromise, pollFallback()]) + cleanupSubscriptionWait()🤖 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 `@packages/envio/src/sources/SourceManager.res` around lines 346 - 383, The subscription resolver pushed into sourceState.pendingHeightResolvers inside waitForSubscription can be left orphaned if pollFallback wins the Promise.race; fix this by registering the resolver in a local variable when creating the Promise (inside waitForSubscription), pushing it into sourceState.pendingHeightResolvers, and ensuring it is removed from sourceState.pendingHeightResolvers if the promise is never fulfilled (e.g., on cleanup/cancellation when pollFallback completes or before returning from waitForSubscription). Modify waitForSubscription (and its Promise.make resolver registration) so that the resolver is removed from sourceState.pendingHeightResolvers in the finally/cancellation path to avoid unbounded growth of pendingHeightResolvers when pollFallback wins the race.
🤖 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 `@packages/envio/src/sources/SourceManager.res`:
- Around line 346-383: The subscription resolver pushed into
sourceState.pendingHeightResolvers inside waitForSubscription can be left
orphaned if pollFallback wins the Promise.race; fix this by registering the
resolver in a local variable when creating the Promise (inside
waitForSubscription), pushing it into sourceState.pendingHeightResolvers, and
ensuring it is removed from sourceState.pendingHeightResolvers if the promise is
never fulfilled (e.g., on cleanup/cancellation when pollFallback completes or
before returning from waitForSubscription). Modify waitForSubscription (and its
Promise.make resolver registration) so that the resolver is removed from
sourceState.pendingHeightResolvers in the finally/cancellation path to avoid
unbounded growth of pendingHeightResolvers when pollFallback wins the race.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: adb310fc-0e04-430a-ba2e-f1d8b9e921f3
📒 Files selected for processing (2)
packages/envio/src/sources/SourceManager.resscenarios/test_codegen/test/lib_tests/SourceManager_test.res
🚧 Files skipped from review as they are similar to previous changes (1)
- scenarios/test_codegen/test/lib_tests/SourceManager_test.res
) Resolve the pending height resolvers only when the stream reports a strictly higher height. The height stream re-emits the current head on every (re)connect, and waking the wait loop on a height we already know spun the loop and created another uncancelled pollingFallback each time, so N stale re-emits produced N concurrent /height poll loops. This fixes the runaway at the source, so the earlier wait-loop restructure is no longer needed and is reverted here, leaving the jittered fallback trigger. https://claude.ai/code/session_01MCsYorVnvp3AKFtbyaiefw
DZakh
left a comment
DZakh
left a comment
There was a problem hiding this comment.
Nice - I like how the change looks now 👍
2 participants
Summary
Fixes issue #1270 where stale (non-increasing) height events from a reconnecting SSE stream would spawn multiple concurrent REST polling fallbacks, overwhelming the height endpoint.
Key Changes
SourceManager.res: Refactored
getSourceNewHeightto use a single, long-lived polling fallback instead of spawning a new one for each stale height event. The subscription and fallback poller now run concurrently in separate loops that internally wait for a qualifying height, preventing the outer loop from spinning on stale events.nextFallbackDelay()to introduce jitter into the fallback polling interval, preventing thundering herd when multiple indexers fall back simultaneouslynewHeight.contents > initialHeighttostatus.contents !== Donefor clearer semanticswaitForSubscriptionandpollFallback) that race against each otherEnvioApiClient.res (new file): Created a REST client wrapper that tags requests with a
hyperindexUser-Agent header, making them attributable on the server side and consistent with SSE and Rust client behaviorHyperSyncSource.res and HyperFuelSource.res: Updated to use
EnvioApiClient.make()instead ofRest.client()for their height polling endpointsSourceManager_test.res: Added comprehensive test case verifying that a burst of 20 stale SSE heights results in at most 1 concurrent /height poll, confirming the fix prevents unbounded polling proliferation
Implementation Details
The core fix changes the fallback polling from a "fire-and-forget" model (where each stale event could trigger a new poller) to a "single persistent poller" model. Both the subscription waiter and the fallback poller contain their own internal loops that continue until they observe a height greater than
knownHeight, so stale events no longer cause the outergetSourceNewHeightloop to iterate and spawn additional pollers.The jittered fallback delay (
[x/2, 1.5x)with meanx) reduces synchronized polling load across multiple indexers experiencing simultaneous subscription staleness.https://claude.ai/code/session_01MCsYorVnvp3AKFtbyaiefw
Summary by CodeRabbit
Bug Fixes
Chores
Tests