Expose the SubscribeRequest sink for runtime filter updates - #302
Open
senzenn wants to merge 9 commits into
Open
Expose the SubscribeRequest sink for runtime filter updates#302senzenn wants to merge 9 commits into
senzenn wants to merge 9 commits into
Conversation
The subscribe request was built inline in connect. Pull it into build_subscribe_request so there is one place that turns a Filters into a wire request, which is what a second caller will need. from_slot stays out of the builder. It is a one-time start position rather than a steady-state setting, which is why the client library overwrites the field with the live checkpoint on reconnect instead of reusing the configured value, so the initial subscribe sets it at the call site.
Sources hold their subscription for the lifetime of the connection, so a caller whose filter set changes has to drop the connection and reconnect to pick it up, losing messages in the window. Add the seam for changing it in place. SourceTrait gains two defaulted methods: supports_filter_updates, and connect_with_filter_updates which ignores the receiver and defers to connect. Defaults rather than a signature change on connect keeps all 17 existing implementations compiling untouched, 11 of which are test mocks. The builder creates the channel and Runtime hands out the sending half via filter_updates, gated on the source advertising support so a caller wiring updates to a source that ignores them finds out at the call site. The runtime drops its own copy of the sender before running, otherwise a source waiting on updates would wait forever on a channel nobody could send to. The payload is Filters rather than SubscribeRequest. The conversion to a wire request leaves commitment and from_slot unset and the gRPC source fills them from config afterwards, so a raw request would bypass that and let the subscription drift from the registered parsers.
subscribe_with_request returns a sink alongside the stream, and the source bound it to _sub_tx and dropped it. Keep it and drive it from the update channel, so the read loop becomes a select over the stream and the receiver. Both trait methods funnel into one private run, so there is a single connection path rather than two copies of it. A rejected send is held and retried rather than logged and dropped. Auto-reconnect is on by default and the client swallows recoverable stream errors to reconnect behind our back, so a send failing mid-reconnect never surfaces on the stream. Worse, the sink records a request into its reconnect state only after a successful send, so the reconnect would come back with the previous filters while the caller's send had already returned Ok. The stream producing again is the signal that the reconnect landed and the sink has a live sender, so that is when a held set goes out. Also spell out in build_subscribe_request why from_slot must not ride along on an update: geyser reads it as a replay request and either replays every slot since or ends the stream when it has no replay buffer, and richat rejects the request outright when the set contains blocks. The update branch retires itself once the sending half is gone, since a closed receiver is ready on every poll and would otherwise spin the loop.
A mock source that records the filter sets it receives, covering the four cases the plumbing can get wrong: a set sent through the handle reaches the source, an unsupporting source hands out no sender, the sender is handed out at most once, and a runtime whose handle was never taken still runs to completion. The last one is a regression guard. try_run_async consumes the runtime but the unclaimed sender used to stay alive inside it for the whole scope, so a source awaiting recv waited on a channel that could never produce. It hangs without the explicit drop. Tests share one static for the recorded IDs and libtest runs them on separate threads, so each records under its own parser ID and asserts on that alone rather than clearing and comparing the whole vector.
Document how to take the handle before the runtime is consumed, that each set replaces the whole subscription rather than adding to it, and that keys are parser IDs so an unregistered one changes what the server sends and then gets discarded at trace level. Say plainly that the feature depends on the provider honouring mid-stream requests. Both yellowstone-grpc-geyser and richat apply them in their source, but a deployment can sit behind infrastructure that does not forward them.
A server that refuses a filter set answers on the stream, not the sink, and the codes it uses are outside the client's recoverable set, so the stream ends and the runtime exits. That was already true before this branch for the initial subscribe, but sending sets mid-stream makes it reachable long after startup, where the cause is far less obvious. Nothing can un-send the request, so record how many updates went out on the connection and report it alongside the status. An operator seeing a run stop on InvalidArgument can then tell a rejected update apart from an unrelated server error. Document the same on the accessor and in the README.
The server applies a new set promptly, but a consumer sees it only once whatever it has already queued drains, so the wait is a property of how far behind the pipeline is rather than of the update itself. Updates matching the new set arrive stale by that margin and settle to real time once the backlog clears. Say that plainly on the accessor and in the README, because a caller who reads the delay the other way will size the wrong problem.
…raffic Review caught that a set held after a sink rejection was only retried inside the arm handling a successful stream item, which assumes the connection keeps producing. It need not. The filter being replaced may match nothing, and the client swallows its own keepalive messages rather than yielding them, so there are live connections where the retry site is never reached and the set is stranded for the rest of the connection. Retry on a five second tick instead, gated on something actually being held so the branch is disabled in the normal case. Log when a connection ends with a set still unsent, since the earlier warning promised a retry that then never happened. Also correct three things review found stale or wrong. The channel-depth doc said sets are never coalesced, which stopped being true when a newer set started superseding a held one. The accessor doc said a refused set ends the run, understating it: run and run_async treat that error as fatal and exit the process. And the delivery test built its payload from Prefilter::default(), which converts to an entirely empty request, so it would have passed on a payload that says nothing on the wire.
Adding the retry timer left the same send-and-count block at three sites and made the copy in the stream arm redundant, since the timer already covers every case. Drop that one so the per-update path carries no filter work at all, and let the send helper own the success count so the two remaining sites are a single line each. Reset the timer when a set is newly held. An interval's first tick is immediate, so without it a rejected set retried at once, inside the same reconnect window that had just rejected it.
senzenn
marked this pull request as draft
September 1, 2026 12:09
senzenn
marked this pull request as ready for review
September 2, 2026 01:58
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.
Summary
Closes #286.
subscribe_with_requestreturns a sink alongside the update stream. The gRPC source bound it to_sub_txand threw it away, so a subscription's filter set was fixed for the life of the connection. Changing filters meant dropping the connection and reconnecting, which loses messages in the window.This keeps the sink.
Runtime::filter_updates()hands the sending half to the caller, and the source applies each set it receives to the live subscription.Approach
Two new
SourceTraitmethods with defaults, rather than changingconnect's signature. There are 17 implementations, 11 of them test mocks, and defaults leave all of them alone.The channel carries
Filters, notSubscribeRequest. Converting to a wire request leavescommitmentandfrom_slotunset and the source fills them from config afterwards, so a raw request would skip that and let the subscription drift from the registered parsers.from_slotonly goes on the initial subscribe. Servers read it as a replay request, so repeating it on an update replays the gap or ends the stream.Fumarole discards its sink the same way at
yellowstone-fumarole-source/src/lib.rs:137. Separate PR, now that the seam exists.Testing
cargo test --workspace, clippy with-Dwarningsandcargo fmt --checkall pass, and each commit builds on its own. New tests cover the request builder and a filter set reaching a source through the channel.Notes for reviewers
run()orrun_async()that is fatal and exits the process.self.pipelines.filters()inruntime/src/lib.rs), so a key matching no pipeline still changes what the server sends while the runtime discards everything it produces.