[streaming-quote-api] Stream per-solver results from the price competition (1/3)#4467
[streaming-quote-api] Stream per-solver results from the price competition (1/3)#4467squadgazzz wants to merge 5 commits into
Conversation
- inline estimate_all into the trait impl, elide lifetime to match PriceEstimating - assert arrival order in the streaming test so a serial impl would fail - drop em dash in doc comment
squadgazzz
changed the title
squadgazzz
changed the title
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the StreamingPriceEstimating trait and its implementation for CompetitionEstimator, which allows price estimation results to be yielded concurrently as they complete. It also adds unit tests to verify that the stream yields all results in the correct order and properly propagates errors. No critical issues were found, and there is no feedback to provide.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
MartinquaXD
left a comment
MartinquaXD
left a comment
There was a problem hiding this comment.
Looks reasonable to me but I'd like to hold the approval until I red all 3 PRs.
| let futures: FuturesUnordered<BoxFuture<'_, PriceEstimateResult>> = FuturesUnordered::new(); | ||
| for stage in &self.stages { | ||
| for (_name, estimator) in stage { | ||
| futures.push(estimator.estimate(query.clone())); |
There was a problem hiding this comment.
maybe we should return the name with the result for logging/metric purposes when being used in the followup stacked PR?
There was a problem hiding this comment.
Nothing consumes the name yet, and the streaming path doesn't expose per-solver identity. If we add per-estimator metrics there, I'll thread it through then.
| /// by dropping the stream. | ||
| fn estimate_stream(&self, query: Arc<Query>) -> BoxStream<'_, PriceEstimateResult> { | ||
| let futures: FuturesUnordered<BoxFuture<'_, PriceEstimateResult>> = FuturesUnordered::new(); | ||
| for stage in &self.stages { |
There was a problem hiding this comment.
Since we iterate self.stages and call each leaf estimator's estimate directly — it does not go through CompetitionEstimator::estimate. So it bypasses the wrapper layer that does is_reasonable and emit_quote_event - is that intended?
There was a problem hiding this comment.
Intended. Going through produce_results pulls in ranking, early-return, and the per-stage timeout division, which streaming doesn't want. The consumer runs the is_reasonable check instead and drops estimates with zero gas or zero out_amount. I left emit_quote_event out of the primitive for v1, and we can add streaming metrics at the consumer if we want them.
| let mut m = MockPriceEstimating::new(); | ||
| m.expect_estimate().times(1).returning(|_| { | ||
| async { | ||
| sleep(Duration::from_millis(10)).await; |
There was a problem hiding this comment.
shouldnt you do a notify instead? making this one block until the fast stream is "done"
There was a problem hiding this comment.
The sleep only makes the ordering observable. The fast estimator resolves on its first poll with no pending await, so FuturesUnordered yields it before the slow estimator's sleep elapses every time. So, a Notify adds machinery without a stronger guarantee.
|
This pull request has been marked as stale because it has been inactive a while. Please update this pull request or it will be automatically closed. |
…ighten error-passthrough test
MartinquaXD
left a comment
MartinquaXD
left a comment
There was a problem hiding this comment.
Follow up PRs look solid so this change seems like a good way to go.
4 participants
Description
First PR for the streaming quote API (#4456). It adds the one capability the streaming endpoint needs from the pricing layer: the ability for the price-estimation competition to emit each solver's result as it arrives, rather than only the single best one once the whole competition finishes. Purely additive. Nothing consumes it yet, the next PR in the stack does.
Changes
StreamingPriceEstimatingtrait next toPriceEstimating.CompetitionEstimator: run every estimator concurrently and yield each result as it completes, with no ranking and no early return. Verification is unaffected, theverifiedflag still rides on each estimate.How to test
New unit tests.
Related issues
Part of #4456