[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 3 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())); | ||
| } | ||
| } | ||
| futures.boxed() |
There was a problem hiding this comment.
It's possible to turn this into a single chain:
self.stages
.iter()
.flatten()
.map(|(_name, estimator)| estimator.estimate(query.clone()))
.collect::<FuturesUnordered<_>>()
.boxed()Feel free to ignore if you prefer the current logic.
| ("ok".to_owned(), Arc::new(ok)), | ||
| ("err".to_owned(), Arc::new(err)), |
There was a problem hiding this comment.
This test could give slightly stronger guarantees that an error doesn't abort the entire stream if you emit the ok response after a tiny sleep.
Having the ordered results can also make the assertion logic at the end a bit easier to read.
| 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?
| /// 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?
3 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