mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-26 15:40:18 +00:00
* Add automated QA: tool schema validator, feature-flag CI matrix, Docker build P0 items from the automated QA plan (#352): - Add validate_tool_schema() that checks OpenAI strict-mode rules (type: object, required keys in properties, nested object/array recursion) with 10 unit tests and 6 integration tests covering all core built-in tools - CI test matrix now runs with --all-features, default features, and --no-default-features --features libsql to catch dead code behind wrong cfg gates - CI clippy now runs the same 3-feature matrix with --all flags - Docker build job added to catch missing files in Dockerfile Co-Authored-By: Claude Opus 4.6 <[email protected]> * Add P1 automated QA tests and fix LeakDetector prefix shadowing bug P1 test coverage: config round-trip (settings + bootstrap), shell tool arg handling, safety adversarial tests (sanitizer, leak detector, allowlist), turn persistence (conversations, metadata, pagination, jobs), and a clippy fix for libsql-only builds. Fixed a real bug where AhoCorasick non-overlapping prefix iteration caused shorter prefixes (e.g. "sk-") to shadow longer ones (e.g. "sk-ant-api"), preventing Anthropic API key and SSH private key detection. Co-Authored-By: Claude Opus 4.6 <[email protected]> * Add P2 automated QA tests: chaos, lifecycle, collision, and recovery Cover all P2 items from the automated QA plan: - Circuit breaker chaos tests (hanging provider, rapid cycles, mixed errors) - Failover chaos tests (hanging failover, all-fail, tools path, single provider) - Value estimator boundary tests (negative cost, zero price, zero earnings) - Context length recovery test (ContextLengthExceeded -> compact -> retry) - WASM channel lifecycle tests (write/commit/read round-trip, namespace isolation) - Extension registry collision tests (same-name different-kind coexistence) - Extension filesystem collision tests (separate dirs, detect_kind priority) Co-Authored-By: Claude Opus 4.6 <[email protected]> * Add P3 concurrent stress tests for ContextManager and SessionManager Tests verify thread safety of double-checked locking, TOCTOU prevention, and RwLock-based concurrent access patterns under load. Co-Authored-By: Claude Opus 4.6 <[email protected]> * Add dispatcher loop guard and self-repair stuck job tests Dispatcher: test force_text mechanism prevents infinite tool call loops, verify iteration bound arithmetic guarantees termination for all configs. Self-repair: test stuck job detection, recovery within attempt limits, manual escalation when limit exceeded, graceful degradation without store/builder dependencies. Co-Authored-By: Claude Opus 4.6 <[email protected]> * Add E2E testing infrastructure design doc Python + Playwright framework with mock LLM server for deterministic browser-level testing of the web gateway. Covers connection/auth, chat round-trip with SSE streaming, and skills lifecycle scenarios. Co-Authored-By: Claude Opus 4.6 <[email protected]> * Add E2E testing infrastructure implementation plan 10-task plan covering: scaffolding, mock LLM server, helpers, conftest fixtures, connection/chat/skills test scenarios, CI workflow, README, and integration run. Co-Authored-By: Claude Opus 4.6 <[email protected]> * scaffold: E2E test project with pyproject.toml Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: E2E helpers with DOM selectors and port discovery Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: mock OpenAI-compat LLM server for E2E tests Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: E2E conftest with session fixtures for mock LLM and ironclaw Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: E2E scenario 1 -- connection and tab navigation tests Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: E2E scenario 2 -- chat message round-trip tests Co-Authored-By: Claude Opus 4.6 <[email protected]> * feat: E2E scenario 3 -- skills search, install, remove tests Co-Authored-By: Claude Opus 4.6 <[email protected]> * ci: add weekly E2E test workflow with Playwright Co-Authored-By: Claude Opus 4.6 <[email protected]> * docs: E2E test README with setup and usage instructions Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: E2E test integration fixes from first run - Use temp file DB instead of :memory: (libSQL :memory: doesn't persist tables across execute_batch) - Fix installed skills selector: #skills-list not #installed-skills - Add pytest-timeout to dependencies - Improve skills install/remove test with wait_for instead of fixed sleeps 8 passed, 1 skipped (skills install depends on ClawHub availability) Co-Authored-By: Claude Opus 4.6 <[email protected]> * test: add OpenAI strict-mode schema validator for all built-in tools (QA 1.1) Add src/tools/schema_validator.rs with validate_strict_schema() that checks tool parameter schemas against OpenAI function calling strict-mode rules: type object at top level, required keys in properties, enum type consistency, array items definitions, nested object recursion, and additionalProperties. 17 tests validate all 34+ built-in tool schemas across 5 test groups: - 9 simple tools (echo, time, json, http, shell, file read/write/list/patch) - 4 job tools (create, list, status, cancel) - 4 skill tools (list, search, install, remove) - 13 inline schemas for extension, routine, and complex job tools - 4 memory tool schemas Co-Authored-By: Claude Opus 4.6 <[email protected]> * test: add E2E scenarios for SSE reconnect, HTML injection, and tool approval (QA 3.3/5/6) Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: E2E test reliability for HTML injection and SSE reconnect - HTML injection: test sanitization directly via JS injection instead of depending on full LLM round-trip (avoids intermittent 404 from mock) - SSE reconnect: increase wait times for DB persistence and relax assertion to check total message count after history reload Co-Authored-By: Claude Opus 4.6 <[email protected]> * style: cargo fmt formatting Co-Authored-By: Claude Opus 4.6 <[email protected]> * test: add WASM and MCP tool schema validation tests (QA 1.1) Extends the schema validator with representative WASM tool schemas (weather, HTTP client, batch processor, status), MCP tool schemas (default, file read, SQL query, strict mode), and defect detection tests for common external schema issues (missing type, typo in required, array without items, enum type mismatch). Co-Authored-By: Claude Opus 4.6 <[email protected]> * test: add auth middleware and compaction module tests Auth middleware (8 new tests): valid/invalid bearer tokens, query param fallback, case sensitivity, empty tokens, whitespace handling. Compaction module (16 new tests): truncation strategy, summarize strategy with mock LLM, workspace fallback, format_turns helper, sequential compactions, coherence after compaction, token decrease verification. Co-Authored-By: Claude Opus 4.6 <[email protected]> * test: add config round-trip integration tests (QA 1.2) Test the full bootstrap .env lifecycle: write via the same format as save_bootstrap_env/upsert_bootstrap_var, read back via dotenvy, and assert values match. Covers LLM backend selection, embedding disable flag, onboard completion flag, session token keys, multi-key preservation across upsert, and special characters (spaces, equals, quotes, backslashes, hashes). Co-Authored-By: Claude Opus 4.6 <[email protected]> * test: add value estimator boundary tests and dispatcher loop guard (QA 4.3/4.4) Value estimator (14 new tests): zero/negative prices, large values, negative cost, exact margin boundaries, custom margin configuration. Dispatcher loop guard (2 new tests): verifies the dispatch loop terminates when all tool calls fail (regression guard for PR #252 infinite loop) and when max iterations are reached. Co-Authored-By: Claude Opus 4.6 <[email protected]> * test: add failover edge cases and provider chaos tests (QA 2.6/4.1) Failover edge cases (4 new tests): cooldown at zero nanos, half-open failure reopens circuit, all providers fail gracefully (no panic), single failing provider with cooldown. Provider chaos tests (15 new tests): flakey provider with retries, hanging provider with timeout, garbage provider, circuit breaker trip/recover, failover chain cascading, non-transient error stops chain, full stack integration (retry + failover + circuit breaker). Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: address PR review feedback on QA tests - Fix Bearer auth case-sensitivity per RFC 6750 (auth.rs) - Refactor bootstrap.rs to expose path-parameterized variants so config_round_trip tests call real code instead of reimplementations - Remove deprecated event_loop fixture, use dynamic ports, minimal env, session-scoped browser, and wire HEADED=1 in E2E conftest - Add cross-referencing doc comments between schema validators - Simplify array validation logic in tool.rs - Bump e2e.yml checkout@v4 to @v6 Co-Authored-By: Claude Opus 4.6 <[email protected]> * style: cargo fmt and fix clippy warning in signal.rs Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: improve E2E fixture error reporting and prevent stdin blocking - Add --no-onboard flag to prevent wizard from blocking in CI - Pipe /dev/null to stdin to prevent any stdin reads from hanging - Add RUST_BACKTRACE=1 for crash diagnostics - On server startup timeout, dump stderr to pytest output so CI logs show why the server failed to start Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: set session-scoped event loop for E2E async fixtures pytest-asyncio 1.3.0 defaults asyncio_default_fixture_loop_scope to None (function scope), causing session-scoped async fixtures to be re-evaluated per test function with independent event loops. Each test then independently attempts to start the ironclaw server, times out at 120s, and wastes ~24 minutes of CI before the job is cancelled. Setting asyncio_default_fixture_loop_scope = "session" ensures all session-scoped async fixtures share a single event loop, so the server starts once and is reused across all tests. Also adds -x flag to pytest in CI to stop on first failure instead of running all 19 tests when the fixture is broken. Co-Authored-By: Claude Opus 4.6 <[email protected]> * fix: set test loop scope to session to match fixture loop scope With asyncio_default_fixture_loop_scope=session but asyncio_default_test_loop_scope=function (the default), tests run on a per-function event loop while fixtures produce objects (Playwright pages, browser contexts) on the session event loop. This event loop mismatch causes the test to hang indefinitely awaiting Playwright operations that are bound to the wrong loop. Setting both scopes to "session" ensures a single event loop is shared across all fixtures and tests, eliminating the deadlock. Co-Authored-By: Claude Opus 4.6 <[email protected]> * ci: add roll-up jobs to match branch protection required checks Branch protection expects "Code Style (fmt + clippy)" and "Run Tests" status checks, but only individual job names were reported. Add roll-up jobs that aggregate results and report the expected names. Co-Authored-By: Claude Opus 4.6 <[email protected]> --------- Co-authored-by: Claude Opus 4.6 <[email protected]>
1324 lines
51 KiB
Rust
1324 lines
51 KiB
Rust
//! Multi-provider LLM failover.
|
|
//!
|
|
//! Wraps multiple LlmProvider instances and tries each in sequence
|
|
//! until one succeeds. Transparent to callers --- same LlmProvider trait.
|
|
//!
|
|
//! Providers that fail repeatedly are temporarily placed in cooldown
|
|
//! so subsequent requests skip them, reducing latency when a provider
|
|
//! is known to be down. Cooldown state is lock-free (atomics only).
|
|
|
|
use std::collections::HashMap;
|
|
use std::future::Future;
|
|
use std::sync::Arc;
|
|
use std::sync::Mutex;
|
|
use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering};
|
|
use std::time::{Duration, Instant};
|
|
|
|
use async_trait::async_trait;
|
|
use rust_decimal::Decimal;
|
|
|
|
use crate::error::LlmError;
|
|
use crate::llm::provider::{
|
|
CompletionRequest, CompletionResponse, LlmProvider, ModelMetadata, ToolCompletionRequest,
|
|
ToolCompletionResponse,
|
|
};
|
|
|
|
use crate::llm::retry::is_retryable;
|
|
|
|
/// Configuration for per-provider cooldown behavior.
|
|
///
|
|
/// When a provider accumulates `failure_threshold` consecutive retryable
|
|
/// failures, it enters cooldown for `cooldown_duration`. During cooldown
|
|
/// the provider is skipped (unless *all* providers are in cooldown, in
|
|
/// which case the oldest-cooled one is tried).
|
|
#[derive(Debug, Clone)]
|
|
pub struct CooldownConfig {
|
|
/// How long a provider stays in cooldown after exceeding the threshold.
|
|
pub cooldown_duration: Duration,
|
|
/// Number of consecutive retryable failures before cooldown activates.
|
|
pub failure_threshold: u32,
|
|
}
|
|
|
|
impl Default for CooldownConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
cooldown_duration: Duration::from_secs(300),
|
|
failure_threshold: 3,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Per-provider cooldown state, entirely lock-free.
|
|
///
|
|
/// All atomic operations use `Relaxed` ordering — consistent with the
|
|
/// existing `last_used` field. Stale reads are harmless: the worst case
|
|
/// is one extra attempt against a provider that just entered cooldown.
|
|
struct ProviderCooldown {
|
|
/// Consecutive retryable failures. Reset to 0 on success.
|
|
failure_count: AtomicU32,
|
|
/// Nanoseconds since `epoch` when cooldown was activated.
|
|
/// 0 means the provider is NOT in cooldown.
|
|
cooldown_activated_nanos: AtomicU64,
|
|
}
|
|
|
|
impl ProviderCooldown {
|
|
fn new() -> Self {
|
|
Self {
|
|
failure_count: AtomicU32::new(0),
|
|
cooldown_activated_nanos: AtomicU64::new(0),
|
|
}
|
|
}
|
|
|
|
/// Check whether the provider is currently in cooldown.
|
|
fn is_in_cooldown(&self, now_nanos: u64, cooldown_nanos: u64) -> bool {
|
|
let activated = self.cooldown_activated_nanos.load(Ordering::Relaxed);
|
|
activated != 0 && now_nanos.saturating_sub(activated) < cooldown_nanos
|
|
}
|
|
|
|
/// Record a retryable failure. Returns `true` if the threshold was
|
|
/// just reached (caller should activate cooldown).
|
|
fn record_failure(&self, threshold: u32) -> bool {
|
|
let prev = self.failure_count.fetch_add(1, Ordering::Relaxed);
|
|
prev + 1 >= threshold
|
|
}
|
|
|
|
/// Activate cooldown at the given timestamp.
|
|
fn activate_cooldown(&self, now_nanos: u64) {
|
|
// Ensure 0 remains a safe "not in cooldown" sentinel.
|
|
self.cooldown_activated_nanos
|
|
.store(now_nanos.max(1), Ordering::Relaxed);
|
|
}
|
|
|
|
/// Reset failure count and clear cooldown (called on success).
|
|
fn reset(&self) {
|
|
self.failure_count.store(0, Ordering::Relaxed);
|
|
self.cooldown_activated_nanos.store(0, Ordering::Relaxed);
|
|
}
|
|
}
|
|
|
|
/// An LLM provider that wraps multiple providers and tries each in sequence
|
|
/// on transient failures.
|
|
///
|
|
/// The first provider in the list is the primary. If it fails with a retryable
|
|
/// error, the next provider is tried, and so on. Non-retryable errors
|
|
/// (e.g. `AuthFailed`, `ContextLengthExceeded`) propagate immediately.
|
|
///
|
|
/// Providers that repeatedly fail with retryable errors are temporarily
|
|
/// placed in cooldown and skipped, reducing latency.
|
|
pub struct FailoverProvider {
|
|
providers: Vec<Arc<dyn LlmProvider>>,
|
|
/// Index of the provider that last handled a request successfully.
|
|
/// Used by `model_name()` and `cost_per_token()` so downstream cost
|
|
/// tracking reflects the provider that actually served the request.
|
|
last_used: AtomicUsize,
|
|
/// Per-provider cooldown tracking (same length as `providers`).
|
|
cooldowns: Vec<ProviderCooldown>,
|
|
/// Reference instant for computing elapsed nanos. Shared across all
|
|
/// cooldown timestamps so they are comparable.
|
|
epoch: Instant,
|
|
/// Cooldown configuration.
|
|
cooldown_config: CooldownConfig,
|
|
/// Request-scoped provider index keyed by Tokio task ID.
|
|
///
|
|
/// This allows `effective_model_name()` to report the provider that handled
|
|
/// the *current* request, even when other concurrent requests update
|
|
/// `last_used`.
|
|
provider_for_task: Mutex<HashMap<tokio::task::Id, usize>>,
|
|
}
|
|
|
|
impl FailoverProvider {
|
|
/// Create a new failover provider with default cooldown settings.
|
|
///
|
|
/// Returns an error if `providers` is empty.
|
|
pub fn new(providers: Vec<Arc<dyn LlmProvider>>) -> Result<Self, LlmError> {
|
|
Self::with_cooldown(providers, CooldownConfig::default())
|
|
}
|
|
|
|
/// Create a new failover provider with explicit cooldown configuration.
|
|
///
|
|
/// Returns an error if `providers` is empty.
|
|
pub fn with_cooldown(
|
|
providers: Vec<Arc<dyn LlmProvider>>,
|
|
cooldown_config: CooldownConfig,
|
|
) -> Result<Self, LlmError> {
|
|
if providers.is_empty() {
|
|
return Err(LlmError::RequestFailed {
|
|
provider: "failover".to_string(),
|
|
reason: "FailoverProvider requires at least one provider".to_string(),
|
|
});
|
|
}
|
|
let cooldowns = (0..providers.len())
|
|
.map(|_| ProviderCooldown::new())
|
|
.collect();
|
|
Ok(Self {
|
|
providers,
|
|
last_used: AtomicUsize::new(0),
|
|
cooldowns,
|
|
epoch: Instant::now(),
|
|
cooldown_config,
|
|
provider_for_task: Mutex::new(HashMap::new()),
|
|
})
|
|
}
|
|
|
|
/// Nanoseconds elapsed since `self.epoch`.
|
|
///
|
|
/// Truncates `u128` → `u64` (wraps after ~584 years of continuous
|
|
/// uptime). Acceptable because `epoch` is set at construction time.
|
|
fn now_nanos(&self) -> u64 {
|
|
self.epoch.elapsed().as_nanos() as u64
|
|
}
|
|
|
|
/// Current Tokio task ID if available.
|
|
fn current_task_id() -> Option<tokio::task::Id> {
|
|
tokio::task::try_id()
|
|
}
|
|
|
|
/// Bind the selected provider index to the current task.
|
|
fn bind_provider_to_current_task(&self, provider_idx: usize) {
|
|
let Some(task_id) = Self::current_task_id() else {
|
|
return;
|
|
};
|
|
if let Ok(mut guard) = self.provider_for_task.lock() {
|
|
guard.insert(task_id, provider_idx);
|
|
}
|
|
}
|
|
|
|
/// Take and remove the provider index bound to the current task.
|
|
fn take_bound_provider_for_current_task(&self) -> Option<usize> {
|
|
let task_id = Self::current_task_id()?;
|
|
self.provider_for_task
|
|
.lock()
|
|
.ok()
|
|
.and_then(|mut guard| guard.remove(&task_id))
|
|
}
|
|
|
|
/// Try each provider in sequence until one succeeds or all fail.
|
|
///
|
|
/// Providers in cooldown are skipped unless *all* providers are in
|
|
/// cooldown, in which case the one with the oldest cooldown timestamp
|
|
/// (most likely to have recovered) is tried.
|
|
async fn try_providers<T, F, Fut>(&self, mut call: F) -> Result<(usize, T), LlmError>
|
|
where
|
|
F: FnMut(Arc<dyn LlmProvider>) -> Fut,
|
|
Fut: Future<Output = Result<T, LlmError>>,
|
|
{
|
|
let now_nanos = self.now_nanos();
|
|
let cooldown_nanos = self.cooldown_config.cooldown_duration.as_nanos() as u64;
|
|
|
|
// Partition providers into available and cooled-down.
|
|
let (mut available, cooled_down): (Vec<usize>, Vec<usize>) = (0..self.providers.len())
|
|
.partition(|&i| !self.cooldowns[i].is_in_cooldown(now_nanos, cooldown_nanos));
|
|
|
|
// Log skipped providers.
|
|
for &i in &cooled_down {
|
|
tracing::info!(
|
|
provider = %self.providers[i].model_name(),
|
|
"Skipping provider (in cooldown)"
|
|
);
|
|
}
|
|
|
|
// Never skip ALL providers: if every provider is in cooldown, pick
|
|
// the one with the oldest cooldown activation (most likely recovered).
|
|
if available.is_empty() {
|
|
let oldest = (0..self.providers.len())
|
|
.min_by_key(|&i| {
|
|
self.cooldowns[i]
|
|
.cooldown_activated_nanos
|
|
.load(Ordering::Relaxed)
|
|
})
|
|
.ok_or_else(|| LlmError::RequestFailed {
|
|
provider: "failover".to_string(),
|
|
reason: "FailoverProvider requires at least one provider".to_string(),
|
|
})?;
|
|
tracing::info!(
|
|
provider = %self.providers[oldest].model_name(),
|
|
"All providers in cooldown, trying oldest-cooled provider"
|
|
);
|
|
available.push(oldest);
|
|
}
|
|
|
|
let mut last_error: Option<LlmError> = None;
|
|
|
|
for (pos, &i) in available.iter().enumerate() {
|
|
let provider = &self.providers[i];
|
|
let result = call(Arc::clone(provider)).await;
|
|
match result {
|
|
Ok(response) => {
|
|
self.last_used.store(i, Ordering::Relaxed);
|
|
self.cooldowns[i].reset();
|
|
return Ok((i, response));
|
|
}
|
|
Err(err) => {
|
|
if !is_retryable(&err) {
|
|
return Err(err);
|
|
}
|
|
|
|
// Increment failure count; activate cooldown if threshold reached.
|
|
if self.cooldowns[i].record_failure(self.cooldown_config.failure_threshold) {
|
|
let nanos = self.now_nanos();
|
|
self.cooldowns[i].activate_cooldown(nanos);
|
|
tracing::warn!(
|
|
provider = %provider.model_name(),
|
|
threshold = self.cooldown_config.failure_threshold,
|
|
cooldown_secs = self.cooldown_config.cooldown_duration.as_secs(),
|
|
"Provider entered cooldown after repeated failures"
|
|
);
|
|
}
|
|
|
|
if pos + 1 < available.len() {
|
|
let next_i = available[pos + 1];
|
|
tracing::warn!(
|
|
provider = %provider.model_name(),
|
|
error = %err,
|
|
next_provider = %self.providers[next_i].model_name(),
|
|
"Provider failed with retryable error, trying next provider"
|
|
);
|
|
}
|
|
last_error = Some(err);
|
|
}
|
|
}
|
|
}
|
|
|
|
Err(last_error.unwrap_or_else(|| LlmError::RequestFailed {
|
|
provider: "failover".to_string(),
|
|
reason: "Invariant violated in FailoverProvider: providers were exhausted but no last_error was recorded (this branch should be unreachable; possible causes: no provider attempts were made or `available` was unexpectedly empty).".to_string(),
|
|
}))
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl LlmProvider for FailoverProvider {
|
|
fn model_name(&self) -> &str {
|
|
self.providers[self.last_used.load(Ordering::Relaxed)].model_name()
|
|
}
|
|
|
|
fn cost_per_token(&self) -> (Decimal, Decimal) {
|
|
self.providers[self.last_used.load(Ordering::Relaxed)].cost_per_token()
|
|
}
|
|
|
|
async fn complete(&self, request: CompletionRequest) -> Result<CompletionResponse, LlmError> {
|
|
let (provider_idx, response) = self
|
|
.try_providers(|provider| {
|
|
let req = request.clone();
|
|
async move { provider.complete(req).await }
|
|
})
|
|
.await?;
|
|
self.bind_provider_to_current_task(provider_idx);
|
|
Ok(response)
|
|
}
|
|
|
|
async fn complete_with_tools(
|
|
&self,
|
|
request: ToolCompletionRequest,
|
|
) -> Result<ToolCompletionResponse, LlmError> {
|
|
let (provider_idx, response) = self
|
|
.try_providers(|provider| {
|
|
let req = request.clone();
|
|
async move { provider.complete_with_tools(req).await }
|
|
})
|
|
.await?;
|
|
self.bind_provider_to_current_task(provider_idx);
|
|
Ok(response)
|
|
}
|
|
|
|
fn active_model_name(&self) -> String {
|
|
self.providers[self.last_used.load(Ordering::Relaxed)].active_model_name()
|
|
}
|
|
|
|
fn set_model(&self, model: &str) -> Result<(), LlmError> {
|
|
for provider in &self.providers {
|
|
provider.set_model(model)?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
async fn list_models(&self) -> Result<Vec<String>, LlmError> {
|
|
let mut all_models = Vec::new();
|
|
|
|
for provider in &self.providers {
|
|
match provider.list_models().await {
|
|
Ok(models) => all_models.extend(models),
|
|
Err(err) => {
|
|
tracing::warn!(
|
|
provider = %provider.model_name(),
|
|
error = %err,
|
|
"Failed to list models from provider, skipping"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
all_models.sort();
|
|
all_models.dedup();
|
|
Ok(all_models)
|
|
}
|
|
|
|
async fn model_metadata(&self) -> Result<ModelMetadata, LlmError> {
|
|
self.providers[self.last_used.load(Ordering::Relaxed)]
|
|
.model_metadata()
|
|
.await
|
|
}
|
|
|
|
fn calculate_cost(&self, input_tokens: u32, output_tokens: u32) -> Decimal {
|
|
self.providers[self.last_used.load(Ordering::Relaxed)]
|
|
.calculate_cost(input_tokens, output_tokens)
|
|
}
|
|
|
|
fn effective_model_name(&self, requested_model: Option<&str>) -> String {
|
|
if let Some(provider_idx) = self.take_bound_provider_for_current_task() {
|
|
return self.providers[provider_idx].effective_model_name(requested_model);
|
|
}
|
|
|
|
self.providers[self.last_used.load(Ordering::Relaxed)].effective_model_name(requested_model)
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
use std::sync::{Mutex, RwLock};
|
|
use std::time::Duration;
|
|
|
|
use crate::llm::provider::{CompletionResponse, FinishReason, ToolCompletionResponse};
|
|
|
|
/// A mock LLM provider that returns a predetermined result.
|
|
struct MockProvider {
|
|
name: String,
|
|
active_model: RwLock<String>,
|
|
input_cost: Decimal,
|
|
output_cost: Decimal,
|
|
complete_result: Mutex<Option<Result<CompletionResponse, LlmError>>>,
|
|
tool_complete_result: Mutex<Option<Result<ToolCompletionResponse, LlmError>>>,
|
|
}
|
|
|
|
impl MockProvider {
|
|
fn succeeding(name: &str, content: &str) -> Self {
|
|
Self {
|
|
name: name.to_string(),
|
|
active_model: RwLock::new(name.to_string()),
|
|
input_cost: Decimal::ZERO,
|
|
output_cost: Decimal::ZERO,
|
|
complete_result: Mutex::new(Some(Ok(CompletionResponse {
|
|
content: content.to_string(),
|
|
input_tokens: 10,
|
|
output_tokens: 5,
|
|
finish_reason: FinishReason::Stop,
|
|
}))),
|
|
tool_complete_result: Mutex::new(Some(Ok(ToolCompletionResponse {
|
|
content: Some(content.to_string()),
|
|
tool_calls: vec![],
|
|
input_tokens: 10,
|
|
output_tokens: 5,
|
|
finish_reason: FinishReason::Stop,
|
|
}))),
|
|
}
|
|
}
|
|
|
|
fn succeeding_with_cost(
|
|
name: &str,
|
|
content: &str,
|
|
input_cost: Decimal,
|
|
output_cost: Decimal,
|
|
) -> Self {
|
|
Self {
|
|
input_cost,
|
|
output_cost,
|
|
..Self::succeeding(name, content)
|
|
}
|
|
}
|
|
|
|
fn failing_retryable(name: &str) -> Self {
|
|
Self {
|
|
name: name.to_string(),
|
|
active_model: RwLock::new(name.to_string()),
|
|
input_cost: Decimal::ZERO,
|
|
output_cost: Decimal::ZERO,
|
|
complete_result: Mutex::new(Some(Err(LlmError::RequestFailed {
|
|
provider: name.to_string(),
|
|
reason: "server error".to_string(),
|
|
}))),
|
|
tool_complete_result: Mutex::new(Some(Err(LlmError::RequestFailed {
|
|
provider: name.to_string(),
|
|
reason: "server error".to_string(),
|
|
}))),
|
|
}
|
|
}
|
|
|
|
fn failing_non_retryable(name: &str) -> Self {
|
|
Self {
|
|
name: name.to_string(),
|
|
active_model: RwLock::new(name.to_string()),
|
|
input_cost: Decimal::ZERO,
|
|
output_cost: Decimal::ZERO,
|
|
complete_result: Mutex::new(Some(Err(LlmError::AuthFailed {
|
|
provider: name.to_string(),
|
|
}))),
|
|
tool_complete_result: Mutex::new(Some(Err(LlmError::AuthFailed {
|
|
provider: name.to_string(),
|
|
}))),
|
|
}
|
|
}
|
|
|
|
fn failing_rate_limited(name: &str) -> Self {
|
|
Self {
|
|
name: name.to_string(),
|
|
active_model: RwLock::new(name.to_string()),
|
|
input_cost: Decimal::ZERO,
|
|
output_cost: Decimal::ZERO,
|
|
complete_result: Mutex::new(Some(Err(LlmError::RateLimited {
|
|
provider: name.to_string(),
|
|
retry_after: Some(Duration::from_secs(30)),
|
|
}))),
|
|
tool_complete_result: Mutex::new(Some(Err(LlmError::RateLimited {
|
|
provider: name.to_string(),
|
|
retry_after: Some(Duration::from_secs(30)),
|
|
}))),
|
|
}
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl LlmProvider for MockProvider {
|
|
fn model_name(&self) -> &str {
|
|
&self.name
|
|
}
|
|
|
|
fn cost_per_token(&self) -> (Decimal, Decimal) {
|
|
(self.input_cost, self.output_cost)
|
|
}
|
|
|
|
async fn complete(
|
|
&self,
|
|
_request: CompletionRequest,
|
|
) -> Result<CompletionResponse, LlmError> {
|
|
self.complete_result
|
|
.lock()
|
|
.unwrap()
|
|
.take()
|
|
.expect("MockProvider::complete called more than once")
|
|
}
|
|
|
|
async fn complete_with_tools(
|
|
&self,
|
|
_request: ToolCompletionRequest,
|
|
) -> Result<ToolCompletionResponse, LlmError> {
|
|
self.tool_complete_result
|
|
.lock()
|
|
.unwrap()
|
|
.take()
|
|
.expect("MockProvider::complete_with_tools called more than once")
|
|
}
|
|
|
|
async fn list_models(&self) -> Result<Vec<String>, LlmError> {
|
|
Ok(vec![self.name.clone()])
|
|
}
|
|
|
|
fn active_model_name(&self) -> String {
|
|
self.active_model.read().unwrap().clone()
|
|
}
|
|
|
|
fn set_model(&self, model: &str) -> Result<(), LlmError> {
|
|
*self.active_model.write().unwrap() = model.to_string();
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
fn make_request() -> CompletionRequest {
|
|
CompletionRequest::new(vec![crate::llm::ChatMessage::user("hello")])
|
|
}
|
|
|
|
fn make_tool_request() -> ToolCompletionRequest {
|
|
ToolCompletionRequest::new(vec![crate::llm::ChatMessage::user("hello")], vec![])
|
|
}
|
|
|
|
// Test 1: Primary succeeds, no failover occurs.
|
|
#[tokio::test]
|
|
async fn primary_succeeds_no_failover() {
|
|
let primary = Arc::new(MockProvider::succeeding("primary", "primary response"));
|
|
let fallback = Arc::new(MockProvider::succeeding("fallback", "fallback response"));
|
|
|
|
let failover = FailoverProvider::new(vec![primary, fallback]).unwrap();
|
|
|
|
let response = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(response.content, "primary response");
|
|
}
|
|
|
|
// Test 2: Primary fails with retryable error, fallback succeeds.
|
|
#[tokio::test]
|
|
async fn primary_fails_retryable_fallback_succeeds() {
|
|
let primary = Arc::new(MockProvider::failing_retryable("primary"));
|
|
let fallback = Arc::new(MockProvider::succeeding("fallback", "fallback response"));
|
|
|
|
let failover = FailoverProvider::new(vec![primary, fallback]).unwrap();
|
|
|
|
let response = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(response.content, "fallback response");
|
|
}
|
|
|
|
// Test 3: All providers fail, returns last error.
|
|
#[tokio::test]
|
|
async fn all_providers_fail_returns_last_error() {
|
|
let primary = Arc::new(MockProvider::failing_retryable("primary"));
|
|
let fallback = Arc::new(MockProvider::failing_retryable("fallback"));
|
|
|
|
let failover = FailoverProvider::new(vec![primary, fallback]).unwrap();
|
|
|
|
let err = failover.complete(make_request()).await.unwrap_err();
|
|
match err {
|
|
LlmError::RequestFailed { provider, .. } => {
|
|
assert_eq!(provider, "fallback");
|
|
}
|
|
other => panic!("expected RequestFailed, got: {other:?}"),
|
|
}
|
|
}
|
|
|
|
// Test 4: Non-retryable error fails immediately, no failover.
|
|
#[tokio::test]
|
|
async fn non_retryable_error_fails_immediately() {
|
|
let primary = Arc::new(MockProvider::failing_non_retryable("primary"));
|
|
let fallback = Arc::new(MockProvider::succeeding("fallback", "fallback response"));
|
|
|
|
let failover = FailoverProvider::new(vec![primary, fallback]).unwrap();
|
|
|
|
let err = failover.complete(make_request()).await.unwrap_err();
|
|
match err {
|
|
LlmError::AuthFailed { provider } => {
|
|
assert_eq!(provider, "primary");
|
|
}
|
|
other => panic!("expected AuthFailed, got: {other:?}"),
|
|
}
|
|
}
|
|
|
|
// Test 5: Three providers, first two fail (retryable), third succeeds.
|
|
#[tokio::test]
|
|
async fn three_providers_first_two_fail_third_succeeds() {
|
|
let p1 = Arc::new(MockProvider::failing_retryable("provider-1"));
|
|
let p2 = Arc::new(MockProvider::failing_rate_limited("provider-2"));
|
|
let p3 = Arc::new(MockProvider::succeeding("provider-3", "third time lucky"));
|
|
|
|
let failover = FailoverProvider::new(vec![p1, p2, p3]).unwrap();
|
|
|
|
let response = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(response.content, "third time lucky");
|
|
}
|
|
|
|
// Test: complete_with_tools follows same failover logic.
|
|
#[tokio::test]
|
|
async fn complete_with_tools_failover() {
|
|
let primary = Arc::new(MockProvider::failing_retryable("primary"));
|
|
let fallback = Arc::new(MockProvider::succeeding("fallback", "tools fallback"));
|
|
|
|
let failover = FailoverProvider::new(vec![primary, fallback]).unwrap();
|
|
|
|
let response = failover
|
|
.complete_with_tools(make_tool_request())
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(response.content.as_deref(), Some("tools fallback"));
|
|
}
|
|
|
|
// Test: model_name and cost_per_token reflect the last-used provider.
|
|
#[tokio::test]
|
|
async fn model_name_and_cost_track_last_used_provider() {
|
|
let fallback_cost = Decimal::new(15, 6); // 0.000015
|
|
|
|
let primary = Arc::new(MockProvider::failing_retryable("primary-model"));
|
|
let fallback = Arc::new(MockProvider::succeeding_with_cost(
|
|
"fallback-model",
|
|
"ok",
|
|
fallback_cost,
|
|
fallback_cost,
|
|
));
|
|
|
|
let failover = FailoverProvider::new(vec![primary, fallback]).unwrap();
|
|
|
|
// Before any call, defaults to primary (index 0).
|
|
assert_eq!(failover.model_name(), "primary-model");
|
|
assert_eq!(failover.cost_per_token(), (Decimal::ZERO, Decimal::ZERO));
|
|
|
|
// After failover, should reflect the fallback provider.
|
|
let _ = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(failover.model_name(), "fallback-model");
|
|
assert_eq!(failover.cost_per_token(), (fallback_cost, fallback_cost));
|
|
}
|
|
|
|
// Test: model reporting is request-scoped under concurrent requests.
|
|
#[tokio::test]
|
|
async fn effective_model_name_is_request_scoped_under_concurrency() {
|
|
let config = CooldownConfig {
|
|
cooldown_duration: Duration::from_secs(60),
|
|
failure_threshold: 3,
|
|
};
|
|
let primary = Arc::new(MultiCallMockProvider::fail_then_ok("primary", 1));
|
|
let fallback = Arc::new(MultiCallMockProvider::always_ok("fallback"));
|
|
let failover =
|
|
Arc::new(FailoverProvider::with_cooldown(vec![primary, fallback], config).unwrap());
|
|
|
|
let (first_done_tx, first_done_rx) = tokio::sync::oneshot::channel::<()>();
|
|
let (second_done_tx, second_done_rx) = tokio::sync::oneshot::channel::<()>();
|
|
|
|
let failover_a = Arc::clone(&failover);
|
|
let task_a = tokio::spawn(async move {
|
|
// First request: primary fails once, fallback serves.
|
|
let _ = failover_a.complete(make_request()).await.unwrap();
|
|
let _ = first_done_tx.send(());
|
|
|
|
// Wait until the second request finishes and updates global state.
|
|
let _ = second_done_rx.await;
|
|
failover_a.effective_model_name(None)
|
|
});
|
|
|
|
let failover_b = Arc::clone(&failover);
|
|
let task_b = tokio::spawn(async move {
|
|
let _ = first_done_rx.await;
|
|
// Second request: primary now succeeds.
|
|
let _ = failover_b.complete(make_request()).await.unwrap();
|
|
let model = failover_b.effective_model_name(None);
|
|
let _ = second_done_tx.send(());
|
|
model
|
|
});
|
|
|
|
let model_b = task_b.await.unwrap();
|
|
let model_a = task_a.await.unwrap();
|
|
|
|
assert_eq!(model_a, "fallback");
|
|
assert_eq!(model_b, "primary");
|
|
}
|
|
|
|
// Test: list_models aggregates from all providers.
|
|
#[tokio::test]
|
|
async fn list_models_aggregates_all() {
|
|
let p1 = Arc::new(MockProvider::succeeding("model-a", "ok"));
|
|
let p2 = Arc::new(MockProvider::succeeding("model-b", "ok"));
|
|
|
|
let failover = FailoverProvider::new(vec![p1, p2]).unwrap();
|
|
|
|
let models = failover.list_models().await.unwrap();
|
|
assert!(models.contains(&"model-a".to_string()));
|
|
assert!(models.contains(&"model-b".to_string()));
|
|
}
|
|
|
|
// --- MultiCallMockProvider for cooldown tests ---
|
|
//
|
|
// Unlike `MockProvider` which uses `.take()` (single-use), this mock
|
|
// tracks a call counter and returns errors for the first N calls,
|
|
// then succeeds.
|
|
|
|
struct MultiCallMockProvider {
|
|
name: String,
|
|
/// How many calls should fail before succeeding. 0 = always succeed.
|
|
fail_count: u32,
|
|
/// Atomically tracks how many times `complete` has been called.
|
|
calls: AtomicU32,
|
|
/// If true, failures are non-retryable (AuthFailed).
|
|
non_retryable: bool,
|
|
}
|
|
|
|
impl MultiCallMockProvider {
|
|
/// Always succeeds.
|
|
fn always_ok(name: &str) -> Self {
|
|
Self {
|
|
name: name.to_string(),
|
|
fail_count: 0,
|
|
calls: AtomicU32::new(0),
|
|
non_retryable: false,
|
|
}
|
|
}
|
|
|
|
/// Fails with retryable error for the first `n` calls, then succeeds.
|
|
fn fail_then_ok(name: &str, n: u32) -> Self {
|
|
Self {
|
|
name: name.to_string(),
|
|
fail_count: n,
|
|
calls: AtomicU32::new(0),
|
|
non_retryable: false,
|
|
}
|
|
}
|
|
|
|
/// Always fails with retryable error.
|
|
fn always_fail(name: &str) -> Self {
|
|
Self {
|
|
name: name.to_string(),
|
|
fail_count: u32::MAX,
|
|
calls: AtomicU32::new(0),
|
|
non_retryable: false,
|
|
}
|
|
}
|
|
|
|
/// Always fails with non-retryable error.
|
|
fn always_fail_non_retryable(name: &str) -> Self {
|
|
Self {
|
|
name: name.to_string(),
|
|
fail_count: u32::MAX,
|
|
calls: AtomicU32::new(0),
|
|
non_retryable: true,
|
|
}
|
|
}
|
|
|
|
fn call_count(&self) -> u32 {
|
|
self.calls.load(Ordering::Relaxed)
|
|
}
|
|
}
|
|
|
|
#[async_trait]
|
|
impl LlmProvider for MultiCallMockProvider {
|
|
fn model_name(&self) -> &str {
|
|
&self.name
|
|
}
|
|
|
|
fn cost_per_token(&self) -> (Decimal, Decimal) {
|
|
(Decimal::ZERO, Decimal::ZERO)
|
|
}
|
|
|
|
async fn complete(
|
|
&self,
|
|
_request: CompletionRequest,
|
|
) -> Result<CompletionResponse, LlmError> {
|
|
let n = self.calls.fetch_add(1, Ordering::Relaxed);
|
|
if n < self.fail_count {
|
|
if self.non_retryable {
|
|
return Err(LlmError::AuthFailed {
|
|
provider: self.name.clone(),
|
|
});
|
|
}
|
|
return Err(LlmError::RequestFailed {
|
|
provider: self.name.clone(),
|
|
reason: format!("call {} failed", n),
|
|
});
|
|
}
|
|
Ok(CompletionResponse {
|
|
content: format!("{} ok", self.name),
|
|
input_tokens: 10,
|
|
output_tokens: 5,
|
|
finish_reason: FinishReason::Stop,
|
|
})
|
|
}
|
|
|
|
async fn complete_with_tools(
|
|
&self,
|
|
_request: ToolCompletionRequest,
|
|
) -> Result<ToolCompletionResponse, LlmError> {
|
|
let n = self.calls.fetch_add(1, Ordering::Relaxed);
|
|
if n < self.fail_count {
|
|
if self.non_retryable {
|
|
return Err(LlmError::AuthFailed {
|
|
provider: self.name.clone(),
|
|
});
|
|
}
|
|
return Err(LlmError::RequestFailed {
|
|
provider: self.name.clone(),
|
|
reason: format!("call {} failed", n),
|
|
});
|
|
}
|
|
Ok(ToolCompletionResponse {
|
|
content: Some(format!("{} ok", self.name)),
|
|
tool_calls: vec![],
|
|
input_tokens: 10,
|
|
output_tokens: 5,
|
|
finish_reason: FinishReason::Stop,
|
|
})
|
|
}
|
|
|
|
async fn list_models(&self) -> Result<Vec<String>, LlmError> {
|
|
Ok(vec![self.name.clone()])
|
|
}
|
|
}
|
|
|
|
// --- Cooldown tests ---
|
|
|
|
// Cooldown test 1: Provider enters cooldown after `threshold` consecutive failures.
|
|
#[tokio::test]
|
|
async fn cooldown_activates_after_threshold() {
|
|
let config = CooldownConfig {
|
|
cooldown_duration: Duration::from_secs(300),
|
|
failure_threshold: 2,
|
|
};
|
|
let p1 = Arc::new(MultiCallMockProvider::always_fail("p1"));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_ok("p2"));
|
|
|
|
let failover =
|
|
FailoverProvider::with_cooldown(vec![p1.clone(), p2.clone()], config).unwrap();
|
|
|
|
// Request 1: p1 fails (count=1, below threshold), p2 succeeds.
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
assert_eq!(p1.call_count(), 1);
|
|
|
|
// Request 2: p1 fails again (count=2, reaches threshold → cooldown), p2 succeeds.
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
assert_eq!(p1.call_count(), 2);
|
|
|
|
// Request 3: p1 should be skipped (in cooldown), only p2 called.
|
|
let prev_p1_calls = p1.call_count();
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
// p1 was NOT called again.
|
|
assert_eq!(p1.call_count(), prev_p1_calls);
|
|
}
|
|
|
|
// Cooldown test 2: Cooldown expires after duration, provider is retried.
|
|
#[tokio::test]
|
|
async fn cooldown_expires_after_duration() {
|
|
let config = CooldownConfig {
|
|
cooldown_duration: Duration::from_millis(1),
|
|
failure_threshold: 1,
|
|
};
|
|
// p1 fails once then succeeds (fail_then_ok with n=1 would work,
|
|
// but we use always_fail to prove it's skipped, then swap).
|
|
let p1 = Arc::new(MultiCallMockProvider::fail_then_ok("p1", 2));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_ok("p2"));
|
|
|
|
let failover =
|
|
FailoverProvider::with_cooldown(vec![p1.clone(), p2.clone()], config).unwrap();
|
|
|
|
// Request 1: p1 fails (threshold=1, enters cooldown immediately), p2 succeeds.
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
assert_eq!(p1.call_count(), 1);
|
|
|
|
// Request 2: p1 in cooldown, skipped. Only p2 called.
|
|
// (But cooldown is 1ms, so wait a bit to let it expire.)
|
|
tokio::time::sleep(Duration::from_millis(5)).await;
|
|
|
|
// After sleep, cooldown should have expired. p1 gets tried again.
|
|
// p1 is set to fail 2 times total, so call #2 (index 1) still fails.
|
|
// But it proves p1 was attempted again after cooldown expired.
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(p1.call_count(), 2); // p1 was retried
|
|
assert_eq!(r.content, "p2 ok"); // p2 handled it
|
|
|
|
// Wait again for cooldown to expire, p1 call #3 (index 2) succeeds.
|
|
tokio::time::sleep(Duration::from_millis(5)).await;
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p1 ok");
|
|
assert_eq!(p1.call_count(), 3);
|
|
}
|
|
|
|
// Cooldown test 3: Never skip all providers — oldest-cooled one is tried.
|
|
#[tokio::test]
|
|
async fn never_skip_all_providers() {
|
|
let config = CooldownConfig {
|
|
cooldown_duration: Duration::from_secs(300),
|
|
failure_threshold: 1,
|
|
};
|
|
// Both providers always fail.
|
|
let p1 = Arc::new(MultiCallMockProvider::always_fail("p1"));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_fail("p2"));
|
|
|
|
let failover =
|
|
FailoverProvider::with_cooldown(vec![p1.clone(), p2.clone()], config).unwrap();
|
|
|
|
// Request 1: both tried, both fail, both enter cooldown.
|
|
let _ = failover.complete(make_request()).await;
|
|
assert_eq!(p1.call_count(), 1);
|
|
assert_eq!(p2.call_count(), 1);
|
|
|
|
// Request 2: all in cooldown, but the oldest-cooled one (p1, activated
|
|
// first) should be tried.
|
|
let prev_total = p1.call_count() + p2.call_count();
|
|
let _ = failover.complete(make_request()).await;
|
|
let new_total = p1.call_count() + p2.call_count();
|
|
// Exactly one more call was made (to the oldest-cooled provider).
|
|
assert_eq!(new_total, prev_total + 1);
|
|
}
|
|
|
|
// Cooldown test 4: Success resets failure count so it never reaches threshold.
|
|
//
|
|
// With threshold=3, accumulate 2 failures then succeed. Verify the
|
|
// atomic counter is back to 0 and no cooldown was activated. Then
|
|
// use a second provider pair to show that without the reset, 3
|
|
// consecutive failures DO trigger cooldown (control case).
|
|
#[tokio::test]
|
|
async fn reset_on_success() {
|
|
let config = CooldownConfig {
|
|
cooldown_duration: Duration::from_secs(300),
|
|
failure_threshold: 3,
|
|
};
|
|
// p1 fails for calls 0,1 then succeeds on call 2+.
|
|
let p1 = Arc::new(MultiCallMockProvider::fail_then_ok("p1", 2));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_ok("p2"));
|
|
|
|
let failover =
|
|
FailoverProvider::with_cooldown(vec![p1.clone(), p2.clone()], config.clone()).unwrap();
|
|
|
|
// Request 1: p1 fails (failure_count=1), p2 succeeds.
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
|
|
// Request 2: p1 fails (failure_count=2, still below threshold=3), p2 succeeds.
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
assert_eq!(p1.call_count(), 2);
|
|
|
|
// Request 3: p1 succeeds (call index 2) → counter resets to 0.
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p1 ok");
|
|
assert_eq!(p1.call_count(), 3);
|
|
|
|
// Verify counter was reset to 0 and no cooldown activated.
|
|
let nanos = failover.now_nanos();
|
|
let cooldown_nanos = failover.cooldown_config.cooldown_duration.as_nanos() as u64;
|
|
assert!(!failover.cooldowns[0].is_in_cooldown(nanos, cooldown_nanos));
|
|
assert_eq!(
|
|
failover.cooldowns[0].failure_count.load(Ordering::Relaxed),
|
|
0
|
|
);
|
|
|
|
// Control: without a success in the middle, 3 failures DO trigger cooldown.
|
|
let p3 = Arc::new(MultiCallMockProvider::always_fail("p3"));
|
|
let p4 = Arc::new(MultiCallMockProvider::always_ok("p4"));
|
|
let control =
|
|
FailoverProvider::with_cooldown(vec![p3.clone(), p4.clone()], config).unwrap();
|
|
for _ in 0..3 {
|
|
let _ = control.complete(make_request()).await.unwrap();
|
|
}
|
|
let nanos = control.now_nanos();
|
|
assert!(control.cooldowns[0].is_in_cooldown(nanos, cooldown_nanos));
|
|
}
|
|
|
|
// Cooldown test 5: threshold-1 failures don't trigger cooldown, threshold does.
|
|
#[tokio::test]
|
|
async fn threshold_boundary() {
|
|
let config = CooldownConfig {
|
|
cooldown_duration: Duration::from_secs(300),
|
|
failure_threshold: 3,
|
|
};
|
|
let p1 = Arc::new(MultiCallMockProvider::always_fail("p1"));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_ok("p2"));
|
|
|
|
let failover =
|
|
FailoverProvider::with_cooldown(vec![p1.clone(), p2.clone()], config).unwrap();
|
|
|
|
// 2 requests: p1 fails twice (below threshold of 3), not in cooldown.
|
|
for _ in 0..2 {
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
}
|
|
assert_eq!(p1.call_count(), 2);
|
|
|
|
// p1 should still be available (not in cooldown).
|
|
let nanos = failover.now_nanos();
|
|
let cooldown_nanos = failover.cooldown_config.cooldown_duration.as_nanos() as u64;
|
|
assert!(!failover.cooldowns[0].is_in_cooldown(nanos, cooldown_nanos));
|
|
|
|
// 3rd request: p1 fails → reaches threshold → enters cooldown.
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
assert_eq!(p1.call_count(), 3);
|
|
|
|
let nanos = failover.now_nanos();
|
|
assert!(failover.cooldowns[0].is_in_cooldown(nanos, cooldown_nanos));
|
|
|
|
// 4th request: p1 should be skipped.
|
|
let prev = p1.call_count();
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
assert_eq!(p1.call_count(), prev); // not called
|
|
}
|
|
|
|
// Cooldown test 6: Non-retryable error returns immediately, no failure bump.
|
|
#[tokio::test]
|
|
async fn non_retryable_does_not_increment_cooldown() {
|
|
let config = CooldownConfig {
|
|
cooldown_duration: Duration::from_secs(300),
|
|
failure_threshold: 1,
|
|
};
|
|
let p1 = Arc::new(MultiCallMockProvider::always_fail_non_retryable("p1"));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_ok("p2"));
|
|
|
|
let failover =
|
|
FailoverProvider::with_cooldown(vec![p1.clone(), p2.clone()], config).unwrap();
|
|
|
|
// Non-retryable error should return immediately.
|
|
let err = failover.complete(make_request()).await.unwrap_err();
|
|
assert!(matches!(err, LlmError::AuthFailed { .. }));
|
|
assert_eq!(p1.call_count(), 1);
|
|
// p2 should NOT have been called (non-retryable = no failover).
|
|
assert_eq!(p2.call_count(), 0);
|
|
|
|
// p1 should NOT be in cooldown (non-retryable doesn't bump count).
|
|
let nanos = failover.now_nanos();
|
|
let cooldown_nanos = failover.cooldown_config.cooldown_duration.as_nanos() as u64;
|
|
assert!(!failover.cooldowns[0].is_in_cooldown(nanos, cooldown_nanos));
|
|
}
|
|
|
|
// Cooldown test 7: Three providers, first in cooldown, second/third available.
|
|
#[tokio::test]
|
|
async fn three_providers_mixed_cooldown() {
|
|
let config = CooldownConfig {
|
|
cooldown_duration: Duration::from_secs(300),
|
|
failure_threshold: 1,
|
|
};
|
|
let p1 = Arc::new(MultiCallMockProvider::always_fail("p1"));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_ok("p2"));
|
|
let p3 = Arc::new(MultiCallMockProvider::always_ok("p3"));
|
|
|
|
let failover =
|
|
FailoverProvider::with_cooldown(vec![p1.clone(), p2.clone(), p3.clone()], config)
|
|
.unwrap();
|
|
|
|
// Request 1: p1 fails → enters cooldown (threshold=1), p2 succeeds.
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
assert_eq!(p1.call_count(), 1);
|
|
|
|
// Request 2: p1 skipped (cooldown), p2 and p3 available.
|
|
let prev = p1.call_count();
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2 ok");
|
|
assert_eq!(p1.call_count(), prev); // p1 skipped
|
|
}
|
|
|
|
// Test: is_retryable correctly classifies errors.
|
|
#[test]
|
|
fn retryable_classification() {
|
|
// Retryable
|
|
assert!(is_retryable(&LlmError::RequestFailed {
|
|
provider: "p".into(),
|
|
reason: "err".into(),
|
|
}));
|
|
assert!(is_retryable(&LlmError::RateLimited {
|
|
provider: "p".into(),
|
|
retry_after: None,
|
|
}));
|
|
assert!(is_retryable(&LlmError::InvalidResponse {
|
|
provider: "p".into(),
|
|
reason: "bad json".into(),
|
|
}));
|
|
assert!(is_retryable(&LlmError::SessionRenewalFailed {
|
|
provider: "p".into(),
|
|
reason: "timeout".into(),
|
|
}));
|
|
assert!(is_retryable(&LlmError::Io(std::io::Error::new(
|
|
std::io::ErrorKind::ConnectionReset,
|
|
"reset"
|
|
))));
|
|
|
|
// Non-retryable
|
|
assert!(!is_retryable(&LlmError::AuthFailed {
|
|
provider: "p".into(),
|
|
}));
|
|
assert!(!is_retryable(&LlmError::SessionExpired {
|
|
provider: "p".into(),
|
|
}));
|
|
assert!(!is_retryable(&LlmError::ContextLengthExceeded {
|
|
used: 100_000,
|
|
limit: 50_000,
|
|
}));
|
|
assert!(!is_retryable(&LlmError::ModelNotAvailable {
|
|
provider: "p".into(),
|
|
model: "m".into(),
|
|
}));
|
|
}
|
|
|
|
// Test: empty providers list returns error (not panic).
|
|
#[test]
|
|
fn empty_providers_returns_error() {
|
|
let result = FailoverProvider::new(vec![]);
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
// Test: activate_cooldown(0) still activates cooldown (sentinel collision fix).
|
|
#[test]
|
|
fn cooldown_at_nanos_zero_still_activates() {
|
|
let cd = ProviderCooldown::new();
|
|
cd.activate_cooldown(0);
|
|
assert!(cd.is_in_cooldown(0, 1000));
|
|
assert_eq!(cd.cooldown_activated_nanos.load(Ordering::Relaxed), 1);
|
|
}
|
|
|
|
// Test: set_model propagates to all providers and active_model_name reflects change.
|
|
#[test]
|
|
fn set_model_propagates_to_all_providers() {
|
|
let p1: Arc<MockProvider> = Arc::new(MockProvider::succeeding("model-a", "ok"));
|
|
let p2: Arc<MockProvider> = Arc::new(MockProvider::succeeding("model-b", "ok"));
|
|
|
|
let failover = FailoverProvider::new(vec![
|
|
Arc::clone(&p1) as Arc<dyn LlmProvider>,
|
|
Arc::clone(&p2) as Arc<dyn LlmProvider>,
|
|
])
|
|
.unwrap();
|
|
|
|
// Before: active_model_name delegates to last_used (index 0 = p1).
|
|
assert_eq!(failover.active_model_name(), "model-a");
|
|
|
|
// Switch model.
|
|
failover.set_model("new-model").unwrap();
|
|
|
|
// Both inner providers should reflect the change.
|
|
assert_eq!(p1.active_model_name(), "new-model");
|
|
assert_eq!(p2.active_model_name(), "new-model");
|
|
|
|
// FailoverProvider itself should report the new model.
|
|
assert_eq!(failover.active_model_name(), "new-model");
|
|
}
|
|
|
|
// === QA Plan P2 - 4.1: Provider chaos tests ===
|
|
|
|
#[tokio::test]
|
|
async fn hanging_provider_failover_to_healthy_one() {
|
|
// When primary hangs, caller can timeout and the secondary should be reachable
|
|
// on a fresh request. The failover itself doesn't timeout individual providers
|
|
// (that's the HTTP client's job), but after the first provider enters cooldown
|
|
// from repeated failures, the failover skips it.
|
|
let p1 = Arc::new(MultiCallMockProvider::always_fail("p1-broken"));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_ok("p2-healthy"));
|
|
|
|
let config = CooldownConfig {
|
|
cooldown_duration: Duration::from_secs(60),
|
|
failure_threshold: 1,
|
|
};
|
|
let failover =
|
|
FailoverProvider::with_cooldown(vec![p1.clone(), p2.clone()], config).unwrap();
|
|
|
|
// First request: p1 fails → cooldown, p2 succeeds.
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2-healthy ok");
|
|
|
|
// Second request: p1 skipped (in cooldown), p2 serves directly.
|
|
let prev_p1 = p1.call_count();
|
|
let r = failover.complete(make_request()).await.unwrap();
|
|
assert_eq!(r.content, "p2-healthy ok");
|
|
assert_eq!(p1.call_count(), prev_p1, "p1 should be skipped in cooldown");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn all_providers_fail_returns_error_not_panic() {
|
|
let p1 = Arc::new(MultiCallMockProvider::always_fail("p1"));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_fail("p2"));
|
|
let p3 = Arc::new(MultiCallMockProvider::always_fail("p3"));
|
|
|
|
let failover = FailoverProvider::new(vec![p1 as Arc<dyn LlmProvider>, p2, p3]).unwrap();
|
|
|
|
// Should return an error, not panic.
|
|
let result = failover.complete(make_request()).await;
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn failover_with_tools_follows_same_path() {
|
|
let p1 = Arc::new(MultiCallMockProvider::always_fail("p1"));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_ok("p2"));
|
|
|
|
let failover = FailoverProvider::new(vec![p1 as Arc<dyn LlmProvider>, p2]).unwrap();
|
|
|
|
let result = failover.complete_with_tools(make_tool_request()).await;
|
|
assert!(result.is_ok());
|
|
assert_eq!(result.unwrap().content.unwrap(), "p2 ok");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn single_provider_failover_still_works() {
|
|
let p1 = Arc::new(MultiCallMockProvider::always_ok("solo"));
|
|
let failover = FailoverProvider::new(vec![p1 as Arc<dyn LlmProvider>]).unwrap();
|
|
|
|
let result = failover.complete(make_request()).await;
|
|
assert!(result.is_ok());
|
|
assert_eq!(result.unwrap().content, "solo ok");
|
|
}
|
|
|
|
// === QA Plan 2.6: Failover edge case tests ===
|
|
|
|
/// When all providers fail with retryable errors, the failover must
|
|
/// return a graceful error (not panic via .unwrap()/.expect()). Verify
|
|
/// the error content includes the last provider's identity.
|
|
#[tokio::test]
|
|
async fn test_failover_all_providers_fail_no_panic() {
|
|
let p1 = Arc::new(MultiCallMockProvider::always_fail("alpha"));
|
|
let p2 = Arc::new(MultiCallMockProvider::always_fail("beta"));
|
|
let p3 = Arc::new(MultiCallMockProvider::always_fail("gamma"));
|
|
|
|
let failover = FailoverProvider::new(vec![
|
|
p1 as Arc<dyn LlmProvider>,
|
|
p2 as Arc<dyn LlmProvider>,
|
|
p3 as Arc<dyn LlmProvider>,
|
|
])
|
|
.unwrap();
|
|
|
|
// All three providers fail. Must return Err, not panic.
|
|
let result = failover.complete(make_request()).await;
|
|
assert!(result.is_err(), "should return error, not panic");
|
|
let err = result.unwrap_err();
|
|
match &err {
|
|
LlmError::RequestFailed { provider, reason } => {
|
|
// The last error should come from the last provider tried.
|
|
assert_eq!(
|
|
provider, "gamma",
|
|
"error should identify the last provider tried"
|
|
);
|
|
assert!(
|
|
reason.contains("failed"),
|
|
"error reason should describe the failure: {}",
|
|
reason
|
|
);
|
|
}
|
|
other => panic!("expected RequestFailed, got: {:?}", other),
|
|
}
|
|
|
|
// Also test complete_with_tools follows the same graceful path.
|
|
let p4 = Arc::new(MultiCallMockProvider::always_fail("delta"));
|
|
let p5 = Arc::new(MultiCallMockProvider::always_fail("epsilon"));
|
|
let failover2 =
|
|
FailoverProvider::new(vec![p4 as Arc<dyn LlmProvider>, p5 as Arc<dyn LlmProvider>])
|
|
.unwrap();
|
|
|
|
let result = failover2.complete_with_tools(make_tool_request()).await;
|
|
assert!(
|
|
result.is_err(),
|
|
"complete_with_tools should also return error, not panic"
|
|
);
|
|
}
|
|
|
|
/// A single provider that always fails with no fallback available.
|
|
/// Verifies the failover returns the error from that provider and
|
|
/// does not panic or produce an "unreachable" invariant violation.
|
|
#[tokio::test]
|
|
async fn test_failover_with_single_provider_failing() {
|
|
let solo = Arc::new(MultiCallMockProvider::always_fail("solo-broken"));
|
|
let failover = FailoverProvider::new(vec![solo.clone() as Arc<dyn LlmProvider>]).unwrap();
|
|
|
|
// First call: should return error from the solo provider.
|
|
let result = failover.complete(make_request()).await;
|
|
assert!(result.is_err());
|
|
match result.unwrap_err() {
|
|
LlmError::RequestFailed { provider, .. } => {
|
|
assert_eq!(provider, "solo-broken");
|
|
}
|
|
other => panic!("expected RequestFailed, got: {:?}", other),
|
|
}
|
|
|
|
// After repeated failures, the single provider enters cooldown.
|
|
// But since it's the only provider, the "never skip all" logic
|
|
// should still try it (as the oldest-cooled provider).
|
|
let config = CooldownConfig {
|
|
cooldown_duration: Duration::from_secs(300),
|
|
failure_threshold: 1,
|
|
};
|
|
let solo2 = Arc::new(MultiCallMockProvider::always_fail("solo-cd"));
|
|
let failover2 =
|
|
FailoverProvider::with_cooldown(vec![solo2.clone() as Arc<dyn LlmProvider>], config)
|
|
.unwrap();
|
|
|
|
// First call: fails, enters cooldown (threshold=1).
|
|
let _ = failover2.complete(make_request()).await;
|
|
assert_eq!(solo2.call_count(), 1);
|
|
|
|
// Second call: provider is in cooldown, but it's the only one,
|
|
// so "never skip all" should try it anyway.
|
|
let result = failover2.complete(make_request()).await;
|
|
assert!(result.is_err(), "should still fail but not panic");
|
|
assert_eq!(
|
|
solo2.call_count(),
|
|
2,
|
|
"sole provider should be retried despite cooldown"
|
|
);
|
|
|
|
// Third call: same behavior, no state corruption.
|
|
let result = failover2.complete(make_request()).await;
|
|
assert!(result.is_err());
|
|
assert_eq!(solo2.call_count(), 3);
|
|
}
|
|
}
|