Flow control for reactive readers - #145
Open
rjhuijsman wants to merge 11 commits into
Open
Conversation
Current Aviator status
This pull request is currently open (not queued). How to mergeTo merge this PR, comment
See the real-time status of this PR on the
Aviator webapp.
Use the Aviator Chrome Extension
to see the status of your PR within GitHub.
|
rjhuijsman
force-pushed
the
rjh.reactive-flow-control
branch
from
September 7, 2026 10:16
ece13d3 to
866d309
Compare
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
Critical mixed-version compatibility failures and insufficient browser flow-control verification remain unresolved.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Pull request overview
Adds acknowledgement-based flow control so slow reactive readers skip stale states and converge on the latest state.
Changes:
- Adds response IDs and acknowledgement transports.
- Implements backend and client flow control.
- Adds Python and browser regression coverage.
File summaries
| File | Description |
|---|---|
tests/reboot/reactivity_test.py |
Tests direct and transitive state skipping. |
tests/reboot/react/test_reactive_reader_flow_control/test.py |
Adds browser flow-control tests, but does not verify that intermediate states are skipped. |
tests/reboot/react/test_reactive_reader_flow_control/test_against_local_envoy.py |
Configures the Envoy integration test. |
tests/reboot/react/test_reactive_reader_flow_control/index.tsx |
Simulates a slow browser consumer. |
tests/reboot/react/test_reactive_reader_flow_control/BUILD.bazel |
Builds and registers the browser test. |
tests/reboot/ping_api_rbt.golden.py |
Updates generated Python fixtures. |
tests/reboot/greeter_rbt.golden.py |
Updates generated Python fixtures. |
tests/reboot/greeter_rbt_react.golden.js |
Updates generated React fixtures. |
tests/reboot/echo_rbt.golden.py |
Updates generated Python fixtures. |
reboot/web/index.ts |
Adds HTTP and WebSocket acknowledgements. |
reboot/templates/reboot.py.j2 |
Generates Python acknowledgement logic, but incorrectly calls the RPC for empty response IDs from older backends. |
reboot/templates/reboot_react.ts.j2 |
Handles aggregated mutation observations. |
reboot/plugin/skills/upgrade/migrations/next/reactive-reader-acknowledgements.md |
Documents deployment compatibility. |
reboot/aio/react.py |
Implements server-side flow control. |
reboot/aio/contexts.py |
Acknowledges transitive reads, but incorrectly calls the RPC for empty response IDs from older backends. |
rbt/v1alpha1/react.proto |
Defines the acknowledgement protocol and RPC. |
Review details
- Files reviewed: 16/17 changed files
- Comments generated: 3
- Review effort level: Balanced
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
rjhuijsman
force-pushed
the
rjh.reactive-flow-control
branch
3 times, most recently
from
September 7, 2026 12:13
81e8fa8 to
1598aef
Compare
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
The critical HTTPS compatibility failure and moderate duplicate client work remain unresolved.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
- Files reviewed: 15/16 changed files
- Comments generated: 2
- Review effort level: Balanced
rjhuijsman
force-pushed
the
rjh.reactive-flow-control
branch
2 times, most recently
from
September 7, 2026 12:33
1727592 to
1391fb2
Compare
rjhuijsman
changed the title
rjhuijsman
commented
Sep 7, 2026
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
Five moderate issues remain involving version-skew recovery, acknowledgement cleanup, and quadratic catch-up processing.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
- Files reviewed: 15/16 changed files
- Comments generated: 5
- Review effort level: Balanced
Contributor
There was a problem hiding this comment.
🟡 Changes recommended
The unresolved critical backlog issue and moderate cancellation-warning issue must be addressed before approval.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
Suppressed comments (1)
tests/reboot/react/test_reactive_reader_flow_control/test.py:63
- This polling loop has no deadline, so a regression that prevents a response leaves the Selenium worker blocked until the entire Bazel target times out, obscuring which round failed. Other browser tests use bounded waits; add a local deadline and fail with the expected count when it expires.
def wait_for_message_count(driver, count: int) -> list[str]:
while len(rendered_messages(driver)) < count:
time.sleep(0.1)
return rendered_messages(driver)
- Files reviewed: 15/16 changed files
- Comments generated: 2
- Review effort level: Balanced
rjhuijsman
force-pushed
the
rjh.reactive-flow-control
branch
from
September 7, 2026 17:59
4acd949 to
6067c58
Compare
rjhuijsman
force-pushed
the
rjh.reactive-flow-control
branch
from
September 10, 2026 13:36
739d9f7 to
35d7cc7
Compare
Gives a reactive read that keeps up with its backend the same updates it would have got without flow control, and says so when one cannot keep up. The flow control added earlier in this series produced one response and then waited to hear that the client had processed it before producing another, so every update cost a full round trip: a client that could easily keep up saw updates at the rate the round trip allowed rather than the rate they happened, and a burst of changes collapsed into a single response even when the client had room for all of them. A client that could not keep up skipped the states it missed silently, leaving a developer to work that out from a screen that updates in jumps. `React.Query` now sends up to `QUERY_RESPONSE_WINDOW` responses before it must hear from its client. Every response carries the ID of its query and its sequence number within that query, and a client names the sequence number of the last response it has fully processed; the backend returns one response's worth of room for each response between what that client last named and this one, so a client that processed several while a single `ContinueQuery` was in flight names only the newest and is credited for all of them. While the window has room the backend sends every state change as it happens. With no room left it holds the response it has ready and keeps asking for states, merging each one into the response it holds: the state it carries becomes the newer one, and the idempotency keys of both are kept, so a mutation the browser is waiting to observe isn't lost along with the response that reported it. When room arrives that one response goes out carrying the latest state, with `skipped_updates` counting the updates it stood in for and `stall_milliseconds` saying how long it waited. A client is therefore at most a window behind however slow it is, and a client that keeps up loses nothing. Past `REPORTABLE_STALL_MILLISECONDS` -- 100ms, about where a person stops experiencing an update as immediate and starts perceiving lag -- both ends say so. The backend logs, at info level: A client of a reactive query to `Greet` skipped 20 updates because it fell 207ms behind and the browser prints the same about itself to its console. Asking for states while there is no room to send them is what makes that count possible, and it costs the backend a run of the reader method per update it then merges away. It also means the hops of a transitive reactive read don't each hold a window's worth: a hop with no room merges rather than letting responses queue up behind it, so a reader reading through another reader ends up one window behind rather than one per hop. The window is a constant, at 10, rather than something we estimate per client; a `TODO` records that. One would behave exactly like the round-trip-per-response protocol this replaces. `QueryRequest.continue_query_response_id` and `QueryResponse.query_response_id` are replaced by `QueryRequest.continue_query_sequence_number` and `QueryResponse.query_id` plus `QueryResponse.sequence_number`. Neither of the old fields was ever released, so their numbers are reused rather than reserved. `continue_query_sequence_number` is `optional` because a heartbeat is an otherwise empty `QueryRequest` and sequence number zero is a real response. `client_continues_query` still decides whether any of this applies, so a client from before continuations existed still gets responses as fast as the backend produces them. Both transports share `_windowed_query()`, so the websocket and streaming paths can't drift apart. TESTED: `//tests/reboot:reactivity_test_py` and `//tests/reboot/react/test_reactive_reader_flow_control:test_against_local_envoy_py` now assert what a window guarantees -- at most a window's worth of responses on the way to the latest state -- rather than one response per continuation, and the browser test renders exactly 11 of the 31 states written on both the websocket and the fetch transport. `reactivity_test_py` also gains `test_reports_a_stalled_client`, and the browser test asserts the console message through Chrome's own log. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TQxdTEi8qsBfxCcuLxrNmQ
An un-annotated parameter turns `mypy` off for the whole function body it belongs to, so leaving one out costs more than the annotation would have. Nothing in the repo said to write them, and review has had to ask for them one function at a time. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TQxdTEi8qsBfxCcuLxrNmQ
rjhuijsman
force-pushed
the
rjh.reactive-flow-control
branch
from
September 10, 2026 13:48
35d7cc7 to
9e59c2c
Compare
Contributor
Author
|
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
benh
requested changes
Sep 10, 2026
benh
left a comment
benh
left a comment
Contributor
There was a problem hiding this comment.
Lets chat, I'd like to simplify a few things here to account for the upcoming changes in reactive readers and caching.
rjhuijsman
force-pushed
the
rjh.reactive-flow-control
branch
from
September 11, 2026 13:10
bc13e01 to
14d3047
Compare
By splitting the single accumulate-and-send loop into two loops, the logic becomes more linear and easier to read. Simplifying the window helper class also reduces complexity.
To avoid interfering with ongoing work in backend-to-backend call coalescing, this removes the backend-to-backend use of flow control.
The rule this branch adds asks every Python parameter to carry a type, and two of the test helpers it also adds did not: the `ReactStub` shim that stands in for a client from before `ContinueQuery` existed, and the three browser-polling helpers in the flow control test, which annotated everything except the `driver` they are handed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TQxdTEi8qsBfxCcuLxrNmQ
reboot-dev-bot
force-pushed
the
rjh.reactive-flow-control
branch
from
September 11, 2026 22:26
cb99e74 to
8eb2e5e
Compare
benh
requested changes
Sep 13, 2026
| self._responses[task] = asyncio.Future() | ||
| self._responses[task].set_result(response) | ||
|
|
||
| await self._used_response[task].wait() |
Contributor
There was a problem hiding this comment.
Okay great glad to see this gone, but just curious did you confirm with your agent that there is no reason why we were doing that in the first place? I don't think there is but it would be great to get its assessment.
Contributor
There was a problem hiding this comment.
Agent's assessment, as requested. Short version: no reason found for the success-path wait to exist; its one plausible job, on the error path, is kept.
What it was. _used_response made the querier's loop block after publishing each React.Query response until the consuming reader had re-run and called call() again. Two effects: downstream, exactly one reader re-execution per response (no skipping); upstream, the gRPC stream went undrained, so once the HTTP/2 window filled the upstream servicer's yield stalled, which is the "flow control between servers".
Where it came from. It was added in mono a42552c3b "Support errors in React" (2024-01-02), with no rationale beyond the field comment. Right before that commit the loop did exactly what this PR does now: consume as fast as responses arrive, each one replacing the previous future (if self._responses[task].done(): self._responses[task] = asyncio.Future()). So this is a return to the original design rather than a new one.
Reasons considered:
- Error observability — holds, and is kept. That commit added the error branch: publish the exception, wake the reader, wait until the reader has picked it up, then back off and retry. Without the wait a retry could replace the exception future before anyone awaited it (the reader misses a transient error and asyncio logs "Future exception was never retrieved"). The PR still does
clear()/wait()in theexcept BaseExceptionbranch, so this is preserved. - Seeing every intermediate state — no. Nothing promises that; mutations between reader re-runs already coalesce at the source, and reboot-dev/mono#4754 asks for skip-to-latest.
- Idempotency keys / read-your-writes — no. The old loop only looked at
HasField('response')and never readidempotency_keys, so nothing depended on delivering every response. - Upstream back-pressure — the one real thing lost. But it was implicit, FIFO, and only kicked in after a transport-sized backlog, which is exactly the #4754 pathology: a queue forms and the consumer walks it in order. If a fast producer feeding a slow transitive consumer turns out to cost real upstream CPU, the explicit window (
client_continues_query) is the right tool;e47d1b8ctook it off the backend-to-backend path to stay clear of the call-coalescing work, not because it can't apply there. - Cancellation / memory — no difference. Memory is bounded at one response either way (the old code held more, in the HTTP/2 buffer), and the
loop()indirection pluscall.cancel()handle cancellation as they did before 2024.
Comment on lines
+377
to
+392
| if self._continuation is not None and self._continuation.done(): | ||
| await self._continuation | ||
| self._continuation = None | ||
|
|
||
| if self._continuation is None: | ||
| self._continuation = asyncio.ensure_future( | ||
| self._stub.ContinueQuery( | ||
| react_pb2.ContinueQueryRequest( | ||
| query_id=query_response.query_id, | ||
| sequence_number=query_response.sequence_number, | ||
| ), | ||
| # The same metadata ensures we're routed to the | ||
| # same server. | ||
| metadata=self._metadata, | ||
| ) | ||
| ) |
Contributor
There was a problem hiding this comment.
This looks like a bug, consider:
(1) First call to continue_past sets self._continuation then returns.
(2) Second call to continue_past where self._continuation is not yet done so it falls off the end of the function and doesn't do anything.
Maybe this never actually stalls in practice but in the degenerate case I do beleive this could cause an infinite hang.
How about a simple async function that we run that reads from a queue? Or if you want to only send the latest then it can just drain the queue after it gets awoken via not queue.empty() and queue.pop_nowait() and then send the last one? Or some other but still having a long-running asyncio task that is responsible for making the ContinueQuery call.
Comment on lines
+557
to
+573
| if (continuation !== undefined && settled) { | ||
| // Throws if continuing failed, which reconnects the query. | ||
| await continuation; | ||
| continuation = undefined; | ||
| } | ||
|
|
||
| if (continuation === undefined) { | ||
| settled = false; | ||
| continuation = continueQuery({ | ||
| endpoint, | ||
| headers: continueHeaders, | ||
| queryId: response.queryId, | ||
| sequenceNumber: response.sequenceNumber, | ||
| signal, | ||
| }).finally(() => { | ||
| settled = true; | ||
| }); |
Contributor
There was a problem hiding this comment.
This looks like it suffers from the same issue as the Python version of this code.
(1) A response comes in and we call continueQuery.
(2) The next response comes in and continueQuery is not done yet so it gets dropped.
| stall_milliseconds = round( | ||
| (time.monotonic() - stall_start) * 1000 | ||
| ) | ||
| if stall_milliseconds > REPORTABLE_STALL_MILLISECONDS: |
Contributor
There was a problem hiding this comment.
See other comment but tl;dr; lets always log it for now please.
| (time.monotonic() - stall_start) * 1000 | ||
| ) | ||
| if stall_milliseconds > REPORTABLE_STALL_MILLISECONDS: | ||
| logger.info( |
Contributor
There was a problem hiding this comment.
Suggested change
| logger.info( | |
| logger.warning( |
| # | ||
| # TODO: estimate this per client from observed throughput rather than | ||
| # fixing it for everyone. | ||
| QUERY_RESPONSE_WINDOW = 10 |
Contributor
There was a problem hiding this comment.
This seems low, will this stall perfectly healthy network links arbitrarily that could have kept up with each message?
Before this change, `QueryResponse.query_id` signalled "no query to
continue" with an empty string, and two places that waited for a
task to end after cancelling it each spelled the cancel-and-await
dance out by hand. Review asked for the field to be `optional`, so
that its absence is tracked as presence rather than as a sentinel
value, and for both places to reuse `wait_for_tasks`.
- `rbt/v1alpha1/react.proto`: `optional string query_id`. The
Python client checks `HasField('query_id')`; the TypeScript
client checks for `undefined`, which is how protobuf-es renders
an unset `optional` field.
- `QueryContinuations.stop()` and the accumulator cleanup in
`_windowed_query()` go through `wait_for_tasks(..., cancel=True)`.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HfMTSCHtzXaNnhCvzQVcQd
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
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.
Last year, when we played Copenhunt, we found https://github.com/reboot-dev/mono/issues/4754 - reactive readers would fall behind if servers produced updates faster than clients could consume them. It was a major issue for a real Reboot app. We fixed #4754 in #4770, but its fix surfaced a bug that prompted us to revert - the bug it surfaced was later found and fixed by @onelxj.
This PR fixes #4754 once and for all 😬😄. See individual commits for more details.