Implement stream-level fallback for LLM providers#74
Conversation
Previously FallbackProvider::stream delegated to primary and never used the fallback provider. It now peeks the primary's first item; if that item is a retryable error (connection failure, 429, 503, timeout), it switches to the fallback provider's full stream. If primary yields at least one event, subsequent errors propagate unchanged so the existing mc-core stream_with_retry can handle them without corrupting partial consumer state. Built on async-stream (already a workspace dep) rather than adding tokio-stream. Includes three unit tests covering fallback-on-error, primary-success, and non-retryable-error-propagation. https://claude.ai/code/session_01R2n6wKqFkYPvHkwaip8EnJ
|
Important Review skippedReview was skipped due to path filters ⛔ Files ignored due to path filters (1)
CodeRabbit blocks several paths by default. You can override this behavior by explicitly including those paths in the path filters. For example, including ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughImplemented stream-level fallback in Changes
Sequence DiagramsequenceDiagram
participant Caller
participant FallbackProvider
participant PrimaryStream
participant FallbackStream
Caller->>FallbackProvider: stream(request)
activate FallbackProvider
FallbackProvider->>PrimaryStream: poll first item (next_item)
activate PrimaryStream
PrimaryStream-->>FallbackProvider: Result<Event> (ok | err)
deactivate PrimaryStream
alt First item is retryable error
FallbackProvider->>PrimaryStream: stop consuming / drop
FallbackProvider->>FallbackStream: consume and yield entire fallback stream
activate FallbackStream
FallbackStream-->>Caller: yield items until exhaustion
deactivate FallbackStream
else First item is ok or non-retryable error
FallbackProvider->>Caller: yield first item
FallbackProvider->>PrimaryStream: continue yielding remaining primary items
activate PrimaryStream
PrimaryStream-->>Caller: yield remaining items until exhaustion
deactivate PrimaryStream
end
deactivate FallbackProvider
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![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 implements a streaming fallback mechanism for LLM providers in the mc-cli crate. It introduces async-stream and futures-core dependencies to facilitate asynchronous stream handling. The FallbackProvider now attempts to consume the primary stream and, if a retryable error is encountered initially, switches to the fallback provider's stream. Comprehensive unit tests have been added to cover success, retryable failure, and non-retryable failure scenarios. The review feedback suggests adding a warning log when a fallback is triggered to improve production observability.
| Some(Err(e)) if e.is_retryable() => { | ||
| drop(primary_stream); | ||
| let mut fb_stream = fallback.stream(&req); | ||
| while let Some(item) = next_item(&mut fb_stream).await { | ||
| yield item; | ||
| } | ||
| } |
There was a problem hiding this comment.
It would be beneficial to add a log message when a fallback occurs. This improves observability and helps in debugging transient issues with the primary provider in production environments.
Some(Err(e)) if e.is_retryable() => {
tracing::warn!(error = %e, "Primary provider failed with retryable error, switching to fallback");
drop(primary_stream);
let mut fb_stream = fallback.stream(&req);
while let Some(item) = next_item(&mut fb_stream).await {
yield item;
}
}![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (1)
mc/crates/mc-cli/src/provider.rs (1)
363-435: Add one regression test for “retryable error after first primary event.”Current tests don’t explicitly cover: primary yields one event, then hits a retryable error, and fallback must remain unused. Adding this case would lock in the key partial-stream guarantee.
Suggested test case
+ #[tokio::test] + async fn retryable_error_after_first_event_does_not_fallback() { + let primary = Arc::new(MockProvider::new(vec![vec![ + Ok(ProviderEvent::TextDelta("from-primary".into())), + Err(retryable_api_err()), + ]])); + let fallback = Arc::new(MockProvider::new(vec![vec![Ok(ProviderEvent::TextDelta( + "should-not-appear".into(), + ))]])); + + let fp = FallbackProvider { + primary: primary.clone() as Arc<dyn LlmProvider>, + fallback: fallback.clone() as Arc<dyn LlmProvider>, + }; + + let events = collect(fp.stream(&empty_request())).await; + assert_eq!(primary.call_count(), 1); + assert_eq!(fallback.call_count(), 0); + assert!(matches!(events.get(1), Some(Err(e)) if e.is_retryable())); + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@mc/crates/mc-cli/src/provider.rs` around lines 363 - 435, Add a new tokio::test that constructs a primary MockProvider emitting a sequence like [Ok(ProviderEvent::TextDelta("from-primary".into())), Err(retryable_api_err())] and a fallback MockProvider with no expected calls, build a FallbackProvider using those, call collect(fp.stream(&empty_request())).await, then assert the collected text deltas include only "from-primary", primary.call_count() == 1, and fallback.call_count() == 0 to ensure a retryable error after the first primary event does not trigger the fallback.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@mc/crates/mc-cli/src/provider.rs`:
- Around line 363-435: Add a new tokio::test that constructs a primary
MockProvider emitting a sequence like
[Ok(ProviderEvent::TextDelta("from-primary".into())), Err(retryable_api_err())]
and a fallback MockProvider with no expected calls, build a FallbackProvider
using those, call collect(fp.stream(&empty_request())).await, then assert the
collected text deltas include only "from-primary", primary.call_count() == 1,
and fallback.call_count() == 0 to ensure a retryable error after the first
primary event does not trigger the fallback.
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 7a5fd7e8-c5f4-4f7d-a527-11919a7bda2a
⛔ Files ignored due to path filters (1)
mc/Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (2)
mc/crates/mc-cli/Cargo.tomlmc/crates/mc-cli/src/provider.rs
When the primary provider fails with a retryable error and FallbackProvider switches to the secondary, emit a tracing::warn! so operators can observe transient failures in production. https://claude.ai/code/session_01R2n6wKqFkYPvHkwaip8EnJ
Pre-existing transitive advisory via reqwest; cargo-deny advisories check was failing in CI. Minimal patch bump fixes both advisories. https://claude.ai/code/session_01R2n6wKqFkYPvHkwaip8EnJ
|
2 participants
What
Implement proper stream-level fallback logic for the
FallbackProvider. When the primary provider's stream encounters a retryable error on the first item, the fallback provider is now used. Non-retryable errors are propagated immediately without attempting fallback.Why
Previously, the fallback provider was not actually used—the code only attempted the primary provider and ignored the fallback. This PR addresses the TODO comment and implements the intended behavior where transient failures (e.g., 503 Service Unavailable) trigger automatic fallback to a secondary provider.
How
FallbackProvider::stream()to useasync_stream::stream!macro for composable async stream handlingnext_item()helper function to safely poll the next item from a streamasync-streamandfutures-coredependencies to support the implementationChecklist
cargo fmt --all && cargo test --workspace && cargo clippy --workspace --all-targetspasseshttps://claude.ai/code/session_01R2n6wKqFkYPvHkwaip8EnJ
Summary by CodeRabbit
Improvements
Chores