mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-07-30 15:26:14 +00:00
Compare commits
26
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8d3c064b29 | ||
|
|
4ac96ade5f | ||
|
|
180a469399 | ||
|
|
3465a44d0e | ||
|
|
a4259af38f | ||
|
|
0d07dd0cdb | ||
|
|
24015ebec1 | ||
|
|
3bc38937e8 | ||
|
|
9229b87b70 | ||
|
|
cb36189db3 | ||
|
|
d2487f42b0 | ||
|
|
5255fce7b2 | ||
|
|
ee668ee93b | ||
|
|
cf8b990fc7 | ||
|
|
78dd74845b | ||
|
|
e7f4c98475 | ||
|
|
fad792068e | ||
|
|
1e2d130022 | ||
|
|
1e21dce735 | ||
|
|
21ae22abec | ||
|
|
50eabafa57 | ||
|
|
7d829af681 | ||
|
|
9e9bc5bff8 | ||
|
|
6d780ef96d | ||
|
|
b70b94b9b4 | ||
|
|
f1fa7d094f |
+52
-96
@@ -84,93 +84,26 @@ def get_provider_penalty(provider: "BaseUpstreamProvider") -> float:
|
||||
return penalty
|
||||
|
||||
|
||||
def should_prefer_model(
|
||||
candidate_model: "Model",
|
||||
candidate_provider: "BaseUpstreamProvider",
|
||||
current_model: "Model",
|
||||
current_provider: "BaseUpstreamProvider",
|
||||
alias: str,
|
||||
) -> bool:
|
||||
"""Determine if candidate model should replace current model for an alias.
|
||||
|
||||
This is the core decision function for model prioritization. It considers:
|
||||
1. Alias matching quality (exact match vs. canonical slug match)
|
||||
2. Model cost (lower is better)
|
||||
3. Provider penalties (e.g., slight preference against OpenRouter)
|
||||
|
||||
Args:
|
||||
candidate_model: The new model being considered
|
||||
candidate_provider: Provider offering the candidate model
|
||||
current_model: The currently selected model for this alias
|
||||
current_provider: Provider offering the current model
|
||||
alias: The model alias being mapped
|
||||
|
||||
Returns:
|
||||
True if candidate should replace current, False otherwise
|
||||
"""
|
||||
|
||||
def get_base_model_id(model_id: str) -> str:
|
||||
"""Get base model ID by removing provider prefix."""
|
||||
return model_id.split("/", 1)[1] if "/" in model_id else model_id
|
||||
|
||||
def alias_priority(model: "Model") -> int:
|
||||
"""Rank how strong the mapping of alias->model is.
|
||||
|
||||
Highest priority when alias exactly equals the model ID without provider prefix.
|
||||
Next when alias equals canonical slug without prefix. Otherwise lowest.
|
||||
"""
|
||||
model_base = get_base_model_id(model.id)
|
||||
if model_base == alias:
|
||||
return 3
|
||||
if model.canonical_slug:
|
||||
canonical_base = get_base_model_id(model.canonical_slug)
|
||||
if canonical_base == alias:
|
||||
return 2
|
||||
return 1
|
||||
|
||||
candidate_alias_priority = alias_priority(candidate_model)
|
||||
current_alias_priority = alias_priority(current_model)
|
||||
|
||||
# If candidate has better alias match, prefer it regardless of cost
|
||||
if candidate_alias_priority > current_alias_priority:
|
||||
return True
|
||||
|
||||
# If current has better alias match, keep it regardless of cost
|
||||
if current_alias_priority > candidate_alias_priority:
|
||||
return False
|
||||
|
||||
# Same alias priority - compare costs
|
||||
candidate_cost = calculate_model_cost_score(candidate_model)
|
||||
current_cost = calculate_model_cost_score(current_model)
|
||||
|
||||
# Apply provider penalties
|
||||
candidate_adjusted = candidate_cost * get_provider_penalty(candidate_provider)
|
||||
current_adjusted = current_cost * get_provider_penalty(current_provider)
|
||||
|
||||
# Prefer lower adjusted cost
|
||||
should_replace = candidate_adjusted < current_adjusted
|
||||
|
||||
return should_replace
|
||||
|
||||
|
||||
def create_model_mappings(
|
||||
upstreams: list["BaseUpstreamProvider"],
|
||||
overrides_by_id: dict[str, tuple],
|
||||
disabled_model_ids: set[str],
|
||||
) -> tuple[dict[str, "Model"], dict[str, "BaseUpstreamProvider"], dict[str, "Model"]]:
|
||||
) -> tuple[
|
||||
dict[str, "Model"], dict[str, list["BaseUpstreamProvider"]], dict[str, "Model"]
|
||||
]:
|
||||
"""Create optimal model mappings based on cost and provider preferences.
|
||||
|
||||
This is the main entry point for the algorithm. It processes all upstream providers
|
||||
and creates three mappings based on cost optimization:
|
||||
|
||||
1. model_instances: alias -> Model (all model aliases mapped to their Model objects)
|
||||
2. provider_map: alias -> UpstreamProvider (which provider to use for each alias)
|
||||
2. provider_map: alias -> List[UpstreamProvider] (sorted list of providers for each alias)
|
||||
3. unique_models: base_id -> Model (unique models without provider prefixes)
|
||||
|
||||
The algorithm:
|
||||
- Processes non-OpenRouter providers first (they're typically cheaper)
|
||||
- Then processes OpenRouter models (they can still win if cheaper)
|
||||
- For each model alias, uses should_prefer_model() to select the best provider
|
||||
- For each model alias, collects all candidates and sorts them by priority and cost.
|
||||
|
||||
Args:
|
||||
upstreams: List of all upstream provider instances
|
||||
@@ -183,8 +116,7 @@ def create_model_mappings(
|
||||
from .payment.models import _row_to_model
|
||||
from .upstream.helpers import resolve_model_alias
|
||||
|
||||
model_instances: dict[str, "Model"] = {}
|
||||
provider_map: dict[str, "BaseUpstreamProvider"] = {}
|
||||
candidates: dict[str, list[tuple["Model", "BaseUpstreamProvider"]]] = {}
|
||||
unique_models: dict[str, "Model"] = {}
|
||||
|
||||
# Separate OpenRouter from other providers
|
||||
@@ -202,24 +134,14 @@ def create_model_mappings(
|
||||
"""Get base model ID by removing provider prefix."""
|
||||
return model_id.split("/", 1)[1] if "/" in model_id else model_id
|
||||
|
||||
def _maybe_set_alias(
|
||||
def _add_candidate(
|
||||
alias: str, model: "Model", provider: "BaseUpstreamProvider"
|
||||
) -> None:
|
||||
"""Set alias to model/provider if not set or if new model is preferred."""
|
||||
"""Add candidate model/provider for an alias."""
|
||||
alias_lower = alias.lower()
|
||||
existing_model = model_instances.get(alias_lower)
|
||||
if not existing_model:
|
||||
# No existing mapping, set it
|
||||
model_instances[alias_lower] = model
|
||||
provider_map[alias_lower] = provider
|
||||
else:
|
||||
# Check if candidate should replace existing
|
||||
existing_provider = provider_map[alias_lower]
|
||||
if should_prefer_model(
|
||||
model, provider, existing_model, existing_provider, alias
|
||||
):
|
||||
model_instances[alias_lower] = model
|
||||
provider_map[alias_lower] = provider
|
||||
if alias_lower not in candidates:
|
||||
candidates[alias_lower] = []
|
||||
candidates[alias_lower].append((model, provider))
|
||||
|
||||
def process_provider_models(
|
||||
upstream: "BaseUpstreamProvider", is_openrouter: bool = False
|
||||
@@ -266,21 +188,55 @@ def create_model_mappings(
|
||||
|
||||
# Try to set each alias
|
||||
for alias in aliases:
|
||||
_maybe_set_alias(alias, model_to_use, upstream)
|
||||
_add_candidate(alias, model_to_use, upstream)
|
||||
|
||||
# Process non-OpenRouter providers first (they're typically cheaper)
|
||||
# Process non-OpenRouter providers first
|
||||
for upstream in other_upstreams:
|
||||
process_provider_models(upstream, is_openrouter=False)
|
||||
|
||||
# Process OpenRouter last - models only win if they're cheaper or better matched
|
||||
# Process OpenRouter last
|
||||
if openrouter:
|
||||
process_provider_models(openrouter, is_openrouter=True)
|
||||
|
||||
# Log provider distribution
|
||||
# Sort candidates and build final maps
|
||||
model_instances: dict[str, "Model"] = {}
|
||||
provider_map: dict[str, list["BaseUpstreamProvider"]] = {}
|
||||
|
||||
def alias_priority(model: "Model", alias: str) -> int:
|
||||
"""Rank how strong the mapping of alias->model is."""
|
||||
model_base = get_base_model_id(model.id)
|
||||
if model_base == alias:
|
||||
return 3
|
||||
if model.canonical_slug:
|
||||
canonical_base = get_base_model_id(model.canonical_slug)
|
||||
if canonical_base == alias:
|
||||
return 2
|
||||
return 1
|
||||
|
||||
for alias, items in candidates.items():
|
||||
# Sort key: (priority DESC, cost ASC)
|
||||
# Using negative cost for DESC sort overall to keep high priority first
|
||||
def sort_key(item: tuple["Model", "BaseUpstreamProvider"]) -> tuple[int, float]:
|
||||
model, provider = item
|
||||
priority = alias_priority(model, alias)
|
||||
cost = calculate_model_cost_score(model)
|
||||
penalty = get_provider_penalty(provider)
|
||||
adjusted_cost = cost * penalty
|
||||
return (priority, -adjusted_cost)
|
||||
|
||||
items.sort(key=sort_key, reverse=True)
|
||||
|
||||
best_model, best_provider = items[0]
|
||||
model_instances[alias] = best_model
|
||||
provider_map[alias] = [p for _, p in items]
|
||||
|
||||
# Log provider distribution (using top provider for stats)
|
||||
provider_counts: dict[str, int] = {}
|
||||
for provider in provider_map.values():
|
||||
provider_name = getattr(provider, "upstream_name", "unknown")
|
||||
provider_counts[provider_name] = provider_counts.get(provider_name, 0) + 1
|
||||
for providers in provider_map.values():
|
||||
if providers:
|
||||
provider = providers[0]
|
||||
provider_name = getattr(provider, "upstream_name", "unknown")
|
||||
provider_counts[provider_name] = provider_counts.get(provider_name, 0) + 1
|
||||
|
||||
logger.debug(
|
||||
f"Updated model mappings with ({len(unique_models)} unique models and {len(model_instances)} aliases)",
|
||||
|
||||
@@ -6,6 +6,15 @@ from .logging import get_logger
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class UpstreamError(Exception):
|
||||
"""Exception raised when an upstream provider fails."""
|
||||
|
||||
def __init__(self, message: str, status_code: int = 502):
|
||||
self.message = message
|
||||
self.status_code = status_code
|
||||
super().__init__(message)
|
||||
|
||||
|
||||
async def http_exception_handler(request: Request, exc: Exception) -> JSONResponse:
|
||||
"""Handle HTTP exceptions and include request ID in response."""
|
||||
request_id = getattr(request.state, "request_id", "unknown")
|
||||
|
||||
@@ -63,6 +63,11 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
|
||||
|
||||
await reset_all_reserved_balances(session)
|
||||
|
||||
if not s.admin_password:
|
||||
logger.warning(
|
||||
f"Admin password is not set. Visit {s.http_url or 'http://localhost:8000'}/admin to set the password."
|
||||
)
|
||||
|
||||
# Apply app metadata from settings
|
||||
try:
|
||||
app.title = s.name
|
||||
|
||||
@@ -6,7 +6,7 @@ import os
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from pydantic.v1 import BaseModel, BaseSettings, Field, validator
|
||||
from pydantic.v1 import BaseModel, BaseSettings, Field
|
||||
from sqlmodel.ext.asyncio.session import AsyncSession
|
||||
|
||||
|
||||
@@ -37,13 +37,6 @@ class Settings(BaseSettings):
|
||||
|
||||
# Cashu
|
||||
cashu_mints: list[str] = Field(default_factory=list, env="CASHU_MINTS")
|
||||
|
||||
@validator("cashu_mints", pre=True, each_item=True)
|
||||
def normalize_mint_url(cls, v: str) -> str:
|
||||
if isinstance(v, str):
|
||||
return v.rstrip("/")
|
||||
return v
|
||||
|
||||
receive_ln_address: str = Field(default="", env="RECEIVE_LN_ADDRESS")
|
||||
primary_mint: str = Field(default="", env="PRIMARY_MINT_URL")
|
||||
primary_mint_unit: str = Field(default="sat", env="PRIMARY_MINT_UNIT")
|
||||
@@ -70,7 +63,7 @@ class Settings(BaseSettings):
|
||||
cors_origins: list[str] = Field(default_factory=lambda: ["*"], env="CORS_ORIGINS")
|
||||
tor_proxy_url: str = Field(default="socks5://127.0.0.1:9050", env="TOR_PROXY_URL")
|
||||
providers_refresh_interval_seconds: int = Field(
|
||||
default=300, env="PROVIDERS_REFRESH_INTERVAL_SECONDS"
|
||||
default=0, env="PROVIDERS_REFRESH_INTERVAL_SECONDS"
|
||||
)
|
||||
pricing_refresh_interval_seconds: int = Field(
|
||||
default=120, env="PRICING_REFRESH_INTERVAL_SECONDS"
|
||||
|
||||
@@ -6,7 +6,7 @@ from typing import Any
|
||||
|
||||
import httpx
|
||||
import websockets
|
||||
from fastapi import APIRouter
|
||||
from fastapi import APIRouter, HTTPException
|
||||
|
||||
from .core.logging import get_logger
|
||||
from .core.settings import settings
|
||||
@@ -389,6 +389,9 @@ async def get_providers(
|
||||
Return cached providers. If include_json, return provider+health; otherwise provider only.
|
||||
Optional filter by pubkey.
|
||||
"""
|
||||
if settings.providers_refresh_interval_seconds == 0:
|
||||
raise HTTPException(status_code=404, detail="Provider discovery is disabled")
|
||||
|
||||
cache = await get_cache()
|
||||
if not cache:
|
||||
await refresh_providers_cache(pubkey=pubkey)
|
||||
|
||||
@@ -253,6 +253,11 @@ async def list_models(
|
||||
else 1.01,
|
||||
)
|
||||
for r in rows
|
||||
if include_disabled
|
||||
or (
|
||||
r.upstream_provider_id in providers_by_id
|
||||
and providers_by_id[r.upstream_provider_id].enabled
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
@@ -265,6 +270,8 @@ async def get_model_by_id(
|
||||
if not row or not row.enabled:
|
||||
return None
|
||||
provider = await session.get(UpstreamProviderRow, provider_id)
|
||||
if not provider or not provider.enabled:
|
||||
return None
|
||||
provider_fee = provider.provider_fee if provider else 1.01
|
||||
return _row_to_model(row, apply_provider_fee=True, provider_fee=provider_fee)
|
||||
|
||||
|
||||
@@ -79,15 +79,29 @@ async def _fetch_btc_usd_price() -> float:
|
||||
"""Fetch the lowest BTC/USD price from multiple exchanges."""
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
try:
|
||||
prices = await asyncio.gather(
|
||||
_kraken_btc_usd(client),
|
||||
_coinbase_btc_usd(client),
|
||||
_binance_btc_usdt(client),
|
||||
)
|
||||
valid_prices = [price for price in prices if price is not None]
|
||||
tasks = [
|
||||
asyncio.create_task(_kraken_btc_usd(client)),
|
||||
asyncio.create_task(_coinbase_btc_usd(client)),
|
||||
asyncio.create_task(_binance_btc_usdt(client)),
|
||||
]
|
||||
valid_prices: list[float] = []
|
||||
|
||||
for future in asyncio.as_completed(tasks):
|
||||
price = await future
|
||||
if price is not None:
|
||||
valid_prices.append(price)
|
||||
|
||||
if len(valid_prices) >= 2:
|
||||
break
|
||||
|
||||
for task in tasks:
|
||||
if not task.done():
|
||||
task.cancel()
|
||||
|
||||
if not valid_prices:
|
||||
logger.error("No valid BTC prices obtained from any exchange")
|
||||
raise ValueError("Unable to fetch BTC price from any exchange")
|
||||
|
||||
return min(valid_prices)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
|
||||
+178
-76
@@ -16,7 +16,7 @@ from .core.db import (
|
||||
create_session,
|
||||
get_session,
|
||||
)
|
||||
from .core.settings import settings
|
||||
from .core.exceptions import UpstreamError
|
||||
from .payment.helpers import (
|
||||
calculate_discounted_max_cost,
|
||||
check_token_balance,
|
||||
@@ -26,14 +26,15 @@ from .payment.helpers import (
|
||||
from .payment.models import Model
|
||||
from .upstream import BaseUpstreamProvider
|
||||
from .upstream.helpers import init_upstreams
|
||||
from .wallet import deserialize_token_from_string
|
||||
|
||||
logger = get_logger(__name__)
|
||||
proxy_router = APIRouter()
|
||||
|
||||
_upstreams: list[BaseUpstreamProvider] = []
|
||||
_model_instances: dict[str, Model] = {} # All aliases -> Model
|
||||
_provider_map: dict[str, BaseUpstreamProvider] = {} # All aliases -> Provider
|
||||
_provider_map: dict[
|
||||
str, list[BaseUpstreamProvider]
|
||||
] = {} # All aliases -> List[Provider]
|
||||
_unique_models: dict[str, Model] = {} # Unique model.id -> Model (no duplicates)
|
||||
|
||||
|
||||
@@ -70,8 +71,8 @@ def get_model_instance(model_id: str) -> Model | None:
|
||||
return _model_instances.get(model_id.lower())
|
||||
|
||||
|
||||
def get_provider_for_model(model_id: str) -> BaseUpstreamProvider | None:
|
||||
"""Get UpstreamProvider for model ID from global cache."""
|
||||
def get_provider_for_model(model_id: str) -> list[BaseUpstreamProvider] | None:
|
||||
"""Get UpstreamProvider list for model ID from global cache."""
|
||||
return _provider_map.get(model_id.lower())
|
||||
|
||||
|
||||
@@ -98,6 +99,8 @@ async def refresh_model_maps() -> None:
|
||||
disabled_model_ids: set[str] = set()
|
||||
|
||||
for provider in provider_rows:
|
||||
if not provider.enabled:
|
||||
continue
|
||||
for model in provider.models:
|
||||
if model.enabled:
|
||||
overrides_by_id[model.id] = (model, provider.provider_fee)
|
||||
@@ -148,32 +151,14 @@ async def proxy(
|
||||
else:
|
||||
model_id = request_body_dict.get("model", "unknown")
|
||||
|
||||
if "https://testnut.cashu.space" in settings.cashu_mints:
|
||||
try:
|
||||
token_str = None
|
||||
if x_cashu_header := headers.get("x-cashu"):
|
||||
token_str = x_cashu_header
|
||||
elif auth_header := headers.get("authorization"):
|
||||
parts = auth_header.split(" ")
|
||||
if len(parts) > 1 and not parts[1].startswith("sk-"):
|
||||
token_str = parts[1]
|
||||
|
||||
if token_str:
|
||||
token_obj = deserialize_token_from_string(token_str)
|
||||
if token_obj.mint == "https://testnut.cashu.space":
|
||||
model_id = "mock/gpt-420-mock"
|
||||
request_body_dict["model"] = model_id
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
model_obj = get_model_instance(model_id)
|
||||
if not model_obj:
|
||||
return create_error_response(
|
||||
"invalid_model", f"Model '{model_id}' not found", 400, request=request
|
||||
)
|
||||
|
||||
upstream = get_provider_for_model(model_id)
|
||||
if not upstream:
|
||||
upstreams = get_provider_for_model(model_id)
|
||||
if not upstreams:
|
||||
return create_error_response(
|
||||
"invalid_model",
|
||||
f"No provider found for model '{model_id}'",
|
||||
@@ -181,6 +166,10 @@ async def proxy(
|
||||
request=request,
|
||||
)
|
||||
|
||||
# todo figure out cost calculation since fallback provider is usually not the same price
|
||||
# Use first provider for initial checks/cost calculation
|
||||
# primary_upstream = upstreams[0]
|
||||
|
||||
_max_cost_for_model = await get_max_cost_for_model(
|
||||
model=model_id, session=session, model_obj=model_obj
|
||||
)
|
||||
@@ -190,14 +179,31 @@ async def proxy(
|
||||
check_token_balance(headers, request_body_dict, max_cost_for_model)
|
||||
|
||||
if x_cashu := headers.get("x-cashu", None):
|
||||
if is_responses_api:
|
||||
return await upstream.handle_x_cashu_responses(
|
||||
request, x_cashu, path, max_cost_for_model, model_obj
|
||||
)
|
||||
else:
|
||||
return await upstream.handle_x_cashu(
|
||||
request, x_cashu, path, max_cost_for_model, model_obj
|
||||
)
|
||||
last_error = None
|
||||
for i, upstream in enumerate(upstreams):
|
||||
try:
|
||||
if is_responses_api:
|
||||
return await upstream.handle_x_cashu_responses(
|
||||
request, x_cashu, path, max_cost_for_model, model_obj
|
||||
)
|
||||
else:
|
||||
return await upstream.handle_x_cashu(
|
||||
request, x_cashu, path, max_cost_for_model, model_obj
|
||||
)
|
||||
except UpstreamError as e:
|
||||
logger.warning(
|
||||
f"Upstream {upstream.provider_type} failed (x-cashu): {e}"
|
||||
)
|
||||
if i == len(upstreams) - 1:
|
||||
last_error = e
|
||||
continue
|
||||
|
||||
return create_error_response(
|
||||
"upstream_error",
|
||||
str(last_error) if last_error else "All upstreams failed",
|
||||
502,
|
||||
request=request,
|
||||
)
|
||||
|
||||
elif auth := headers.get("authorization", None):
|
||||
key = await get_bearer_token_key(headers, path, session, auth)
|
||||
@@ -212,56 +218,152 @@ async def proxy(
|
||||
)
|
||||
|
||||
logger.debug("Processing unauthenticated GET request", extra={"path": path})
|
||||
headers = upstream.prepare_headers(dict(request.headers))
|
||||
return await upstream.forward_get_request(request, path, headers)
|
||||
|
||||
last_error_response = None
|
||||
for i, upstream in enumerate(upstreams):
|
||||
try:
|
||||
headers = upstream.prepare_headers(dict(request.headers))
|
||||
response = await upstream.forward_get_request(request, path, headers)
|
||||
|
||||
if response.status_code in [502, 429] and i < len(upstreams) - 1:
|
||||
error_message = ""
|
||||
try:
|
||||
if hasattr(response, "body"):
|
||||
body_bytes = response.body
|
||||
data = json.loads(body_bytes)
|
||||
if "error" in data:
|
||||
error_data = data["error"]
|
||||
if isinstance(error_data, dict):
|
||||
error_message = error_data.get("message", "")
|
||||
elif isinstance(error_data, str):
|
||||
error_message = error_data
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
await upstream.on_upstream_error_redirect(
|
||||
response.status_code, error_message
|
||||
)
|
||||
|
||||
logger.warning(
|
||||
f"Upstream {upstream.provider_type} returned {response.status_code} (GET), trying next provider",
|
||||
extra={
|
||||
"status_code": response.status_code,
|
||||
"upstream": upstream.provider_type,
|
||||
},
|
||||
)
|
||||
continue
|
||||
return response
|
||||
except UpstreamError as e:
|
||||
logger.warning(f"Upstream {upstream.provider_type} failed (GET): {e}")
|
||||
if i == len(upstreams) - 1:
|
||||
last_error_response = create_error_response(
|
||||
"upstream_error", str(e), 502, request=request
|
||||
)
|
||||
continue
|
||||
return last_error_response or create_error_response(
|
||||
"upstream_error", "All upstreams failed", 502, request=request
|
||||
)
|
||||
|
||||
if request_body_dict:
|
||||
await pay_for_request(key, max_cost_for_model, session)
|
||||
|
||||
headers = upstream.prepare_headers(dict(request.headers))
|
||||
for i, upstream in enumerate(upstreams):
|
||||
headers = upstream.prepare_headers(dict(request.headers))
|
||||
|
||||
if is_responses_api:
|
||||
response = await upstream.forward_responses_request(
|
||||
request,
|
||||
path,
|
||||
headers,
|
||||
request_body,
|
||||
key,
|
||||
max_cost_for_model,
|
||||
session,
|
||||
model_obj,
|
||||
)
|
||||
else:
|
||||
response = await upstream.forward_request(
|
||||
request,
|
||||
path,
|
||||
headers,
|
||||
request_body,
|
||||
key,
|
||||
max_cost_for_model,
|
||||
session,
|
||||
model_obj,
|
||||
)
|
||||
try:
|
||||
if is_responses_api:
|
||||
response = await upstream.forward_responses_request(
|
||||
request,
|
||||
path,
|
||||
headers,
|
||||
request_body,
|
||||
key,
|
||||
max_cost_for_model,
|
||||
session,
|
||||
model_obj,
|
||||
)
|
||||
else:
|
||||
response = await upstream.forward_request(
|
||||
request,
|
||||
path,
|
||||
headers,
|
||||
request_body,
|
||||
key,
|
||||
max_cost_for_model,
|
||||
session,
|
||||
model_obj,
|
||||
)
|
||||
|
||||
if response.status_code != 200:
|
||||
await revert_pay_for_request(key, session, max_cost_for_model)
|
||||
logger.warning(
|
||||
"Upstream request failed, revert payment",
|
||||
extra={
|
||||
"status_code": response.status_code,
|
||||
"path": path,
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"key_balance": key.balance,
|
||||
"max_cost_for_model": max_cost_for_model,
|
||||
"upstream_headers": response.headers
|
||||
if hasattr(response, "headers")
|
||||
else None,
|
||||
},
|
||||
)
|
||||
# Return the mapped error response generated earlier rather than masking with 502
|
||||
return response
|
||||
if response.status_code != 200:
|
||||
# Check if we should retry (502 Upstream Error or 429 Rate Limit)
|
||||
should_retry = response.status_code in [502, 429, 400, 401, 403, 404]
|
||||
if should_retry and i < len(upstreams) - 1:
|
||||
error_message = ""
|
||||
try:
|
||||
if hasattr(response, "body"):
|
||||
body_bytes = response.body
|
||||
data = json.loads(body_bytes)
|
||||
if "error" in data:
|
||||
error_data = data["error"]
|
||||
if isinstance(error_data, dict):
|
||||
error_message = error_data.get("message", "")
|
||||
elif isinstance(error_data, str):
|
||||
error_message = error_data
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
return response
|
||||
await upstream.on_upstream_error_redirect(
|
||||
response.status_code, error_message
|
||||
)
|
||||
|
||||
logger.warning(
|
||||
f"Upstream {upstream.provider_type} returned {response.status_code}, trying next provider",
|
||||
extra={
|
||||
"status_code": response.status_code,
|
||||
"upstream": upstream.provider_type,
|
||||
},
|
||||
)
|
||||
continue
|
||||
|
||||
# 4xx error (user error), or other non-retryable error, or last provider failed
|
||||
await revert_pay_for_request(key, session, max_cost_for_model)
|
||||
logger.warning(
|
||||
"Upstream request failed, revert payment",
|
||||
extra={
|
||||
"status_code": response.status_code,
|
||||
"path": path,
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"key_balance": key.balance,
|
||||
"max_cost_for_model": max_cost_for_model,
|
||||
"upstream_headers": response.headers
|
||||
if hasattr(response, "headers")
|
||||
else None,
|
||||
},
|
||||
)
|
||||
return response
|
||||
|
||||
return response
|
||||
|
||||
except UpstreamError as e:
|
||||
logger.warning(
|
||||
f"Upstream {upstream.provider_type} failed: {e}",
|
||||
extra={"retry": i < len(upstreams) - 1},
|
||||
)
|
||||
|
||||
# If this was the last provider
|
||||
if i == len(upstreams) - 1:
|
||||
await revert_pay_for_request(key, session, max_cost_for_model)
|
||||
return create_error_response(
|
||||
"upstream_error", str(e), 502, request=request
|
||||
)
|
||||
|
||||
# Otherwise loop continues to next provider
|
||||
continue
|
||||
|
||||
# Should not be reached given logic above
|
||||
return create_error_response(
|
||||
"upstream_error", "All upstreams failed", 502, request=request
|
||||
)
|
||||
|
||||
|
||||
async def get_bearer_token_key(
|
||||
|
||||
@@ -15,6 +15,7 @@ from pydantic import BaseModel
|
||||
from ..auth import adjust_payment_for_tokens
|
||||
from ..core import get_logger
|
||||
from ..core.db import ApiKey, AsyncSession, create_session
|
||||
from ..core.exceptions import UpstreamError
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from ..core.db import UpstreamProviderRow
|
||||
@@ -340,6 +341,20 @@ class BaseUpstreamProvider:
|
||||
message = preview[:500]
|
||||
return message, upstream_code
|
||||
|
||||
async def on_upstream_error_redirect(
|
||||
self, status_code: int, error_message: str
|
||||
) -> None:
|
||||
"""Hook called when the proxy redirects to another provider due to an error.
|
||||
|
||||
Subclasses can implement this to perform actions like disabling the provider
|
||||
if it's out of balance.
|
||||
|
||||
Args:
|
||||
status_code: The HTTP status code returned by the upstream
|
||||
error_message: The error message extracted from the upstream response
|
||||
"""
|
||||
pass
|
||||
|
||||
async def map_upstream_error_response(
|
||||
self, request: Request, path: str, upstream_response: httpx.Response
|
||||
) -> Response:
|
||||
@@ -1148,6 +1163,14 @@ class BaseUpstreamProvider:
|
||||
)
|
||||
|
||||
if response.status_code != 200:
|
||||
if response.status_code >= 500:
|
||||
await response.aclose()
|
||||
await client.aclose()
|
||||
raise UpstreamError(
|
||||
f"Upstream returned status {response.status_code}",
|
||||
status_code=response.status_code,
|
||||
)
|
||||
|
||||
try:
|
||||
mapped_error = await self.map_upstream_error_response(
|
||||
request, path, response
|
||||
@@ -1238,6 +1261,9 @@ class BaseUpstreamProvider:
|
||||
background=background_tasks,
|
||||
)
|
||||
|
||||
except UpstreamError:
|
||||
raise
|
||||
|
||||
except httpx.RequestError as exc:
|
||||
await client.aclose()
|
||||
error_type = type(exc).__name__
|
||||
@@ -1265,9 +1291,7 @@ class BaseUpstreamProvider:
|
||||
else:
|
||||
error_message = f"Error connecting to upstream service: {error_type}"
|
||||
|
||||
return create_error_response(
|
||||
"upstream_error", error_message, 502, request=request
|
||||
)
|
||||
raise UpstreamError(error_message, status_code=502)
|
||||
|
||||
except Exception as exc:
|
||||
await client.aclose()
|
||||
@@ -1380,6 +1404,14 @@ class BaseUpstreamProvider:
|
||||
)
|
||||
|
||||
if response.status_code != 200:
|
||||
if response.status_code >= 500:
|
||||
await response.aclose()
|
||||
await client.aclose()
|
||||
raise UpstreamError(
|
||||
f"Upstream returned status {response.status_code}",
|
||||
status_code=response.status_code,
|
||||
)
|
||||
|
||||
try:
|
||||
mapped_error = await self.map_upstream_error_response(
|
||||
request, path, response
|
||||
@@ -1447,6 +1479,9 @@ class BaseUpstreamProvider:
|
||||
background=background_tasks,
|
||||
)
|
||||
|
||||
except UpstreamError:
|
||||
raise
|
||||
|
||||
except httpx.RequestError as exc:
|
||||
await client.aclose()
|
||||
error_type = type(exc).__name__
|
||||
@@ -1474,9 +1509,7 @@ class BaseUpstreamProvider:
|
||||
else:
|
||||
error_message = f"Error connecting to upstream service: {error_type}"
|
||||
|
||||
return create_error_response(
|
||||
"upstream_error", error_message, 502, request=request
|
||||
)
|
||||
raise UpstreamError(error_message, status_code=502)
|
||||
|
||||
except Exception as exc:
|
||||
await client.aclose()
|
||||
|
||||
@@ -1,265 +0,0 @@
|
||||
import asyncio
|
||||
import json
|
||||
import random
|
||||
from typing import AsyncIterator
|
||||
|
||||
from fastapi import Request
|
||||
from fastapi.responses import Response, StreamingResponse
|
||||
|
||||
from ..core.db import ApiKey, AsyncSession
|
||||
from ..payment.models import Architecture, Model, Pricing
|
||||
from .base import BaseUpstreamProvider
|
||||
|
||||
|
||||
class MockUpstreamProvider(BaseUpstreamProvider):
|
||||
"""Fack Mock Upstream provider specifically for Testing."""
|
||||
|
||||
provider_type = "mock"
|
||||
|
||||
async def forward_request(
|
||||
self,
|
||||
request: Request,
|
||||
path: str,
|
||||
headers: dict,
|
||||
request_body: bytes | None,
|
||||
key: ApiKey,
|
||||
max_cost_for_model: int,
|
||||
session: AsyncSession,
|
||||
model_obj: Model,
|
||||
) -> Response | StreamingResponse:
|
||||
if path.endswith("chat/completions"):
|
||||
is_streaming = False
|
||||
if request_body:
|
||||
request_data = json.loads(request_body)
|
||||
is_streaming = request_data.get("stream", False)
|
||||
|
||||
if is_streaming:
|
||||
|
||||
async def fake_streaming_response(
|
||||
chunk_size: int | None = None,
|
||||
) -> AsyncIterator[bytes]:
|
||||
suffix = random.randint(1000, 9999)
|
||||
req_id = f"gen-mock-stream-{suffix}"
|
||||
created = 1766138895
|
||||
model = "mock/gpt-420-mock"
|
||||
|
||||
def make_chunk(
|
||||
delta: dict,
|
||||
finish_reason: str | None = None,
|
||||
usage: dict | None = None,
|
||||
) -> bytes:
|
||||
chunk = {
|
||||
"id": req_id,
|
||||
"provider": "MockProvider",
|
||||
"model": model,
|
||||
"object": "chat.completion.chunk",
|
||||
"created": created,
|
||||
"choices": [
|
||||
{
|
||||
"index": 0,
|
||||
"delta": delta,
|
||||
"finish_reason": finish_reason,
|
||||
"native_finish_reason": "completed"
|
||||
if finish_reason
|
||||
else None,
|
||||
"logprobs": None,
|
||||
}
|
||||
],
|
||||
}
|
||||
if usage:
|
||||
chunk["usage"] = usage
|
||||
return f"data: {json.dumps(chunk)}\n\n".encode()
|
||||
|
||||
# 1. Initial chunk
|
||||
yield make_chunk({"role": "assistant", "content": ""})
|
||||
await asyncio.sleep(0.02)
|
||||
|
||||
# 2. Reasoning chunks
|
||||
reasoning_tokens = ["Mock", " reason", "ing", "..."]
|
||||
for token in reasoning_tokens:
|
||||
delta = {
|
||||
"role": "assistant",
|
||||
"content": "",
|
||||
"reasoning": token,
|
||||
"reasoning_details": [
|
||||
{
|
||||
"type": "reasoning.summary",
|
||||
"summary": token,
|
||||
"format": "openai-responses-v1",
|
||||
"index": 0,
|
||||
}
|
||||
],
|
||||
}
|
||||
yield make_chunk(delta)
|
||||
await asyncio.sleep(0.03)
|
||||
|
||||
# 3. Content chunks
|
||||
content_tokens = ["This", " is", " a", " mock", " stream", "."]
|
||||
for token in content_tokens:
|
||||
yield make_chunk({"role": "assistant", "content": token})
|
||||
await asyncio.sleep(0.03)
|
||||
|
||||
# 4. Finish chunk
|
||||
yield make_chunk(
|
||||
{"role": "assistant", "content": ""}, finish_reason="stop"
|
||||
)
|
||||
|
||||
# 5. Usage chunk
|
||||
usage_data = {
|
||||
"prompt_tokens": 10,
|
||||
"completion_tokens": 20,
|
||||
"total_tokens": 30,
|
||||
"cost": 0.001,
|
||||
"is_byok": False,
|
||||
"prompt_tokens_details": {
|
||||
"cached_tokens": 0,
|
||||
"audio_tokens": 0,
|
||||
"video_tokens": 0,
|
||||
},
|
||||
"cost_details": {
|
||||
"upstream_inference_cost": None,
|
||||
"upstream_inference_prompt_cost": 0,
|
||||
"upstream_inference_completions_cost": 0.001,
|
||||
},
|
||||
"completion_tokens_details": {
|
||||
"reasoning_tokens": 10,
|
||||
"image_tokens": 0,
|
||||
},
|
||||
}
|
||||
|
||||
usage_chunk = {
|
||||
"id": req_id,
|
||||
"provider": "MockProvider",
|
||||
"model": model,
|
||||
"object": "chat.completion.chunk",
|
||||
"created": created,
|
||||
"choices": [
|
||||
{
|
||||
"index": 0,
|
||||
"delta": {"role": "assistant", "content": ""},
|
||||
"finish_reason": None,
|
||||
"native_finish_reason": None,
|
||||
"logprobs": None,
|
||||
}
|
||||
],
|
||||
"usage": usage_data,
|
||||
}
|
||||
yield f"data: {json.dumps(usage_chunk)}\n\n".encode()
|
||||
|
||||
# 6. DONE
|
||||
yield b"data: [DONE]\n\n"
|
||||
|
||||
# 7. Cost
|
||||
cost_chunk = {
|
||||
"cost": {
|
||||
"base_msats": 0,
|
||||
"input_msats": 2,
|
||||
"output_msats": 10,
|
||||
"total_msats": 12,
|
||||
}
|
||||
}
|
||||
yield f"data: {json.dumps(cost_chunk)}\n\n".encode()
|
||||
|
||||
return StreamingResponse(
|
||||
fake_streaming_response(),
|
||||
200,
|
||||
)
|
||||
|
||||
else:
|
||||
suffix = random.randint(1000, 9999)
|
||||
content_dict = {
|
||||
"id": f"gen-mock-{suffix}",
|
||||
"provider": "MockProvider",
|
||||
"model": "mock/gpt-5-mini",
|
||||
"object": "chat.completion",
|
||||
"created": 1766138655,
|
||||
"choices": [
|
||||
{
|
||||
"logprobs": None,
|
||||
"finish_reason": "length",
|
||||
"native_finish_reason": "max_output_tokens",
|
||||
"index": 0,
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": f"Mock Content {suffix}",
|
||||
"refusal": None,
|
||||
"reasoning": f"Mock Reasoning {suffix}",
|
||||
"reasoning_details": [
|
||||
{
|
||||
"format": "openai-responses-v1",
|
||||
"index": 0,
|
||||
"type": "reasoning.summary",
|
||||
"summary": f"Mock Summary {suffix}",
|
||||
},
|
||||
{
|
||||
"id": f"rs_mock_{suffix}",
|
||||
"format": "openai-responses-v1",
|
||||
"index": 0,
|
||||
"type": "reasoning.encrypted",
|
||||
"data": "mock_encrypted_data",
|
||||
},
|
||||
],
|
||||
},
|
||||
}
|
||||
],
|
||||
"usage": {
|
||||
"prompt_tokens": 10,
|
||||
"completion_tokens": 10,
|
||||
"total_tokens": 20,
|
||||
"cost": 0,
|
||||
"is_byok": False,
|
||||
"prompt_tokens_details": {
|
||||
"cached_tokens": 0,
|
||||
"audio_tokens": 0,
|
||||
"video_tokens": 0,
|
||||
},
|
||||
"cost_details": {
|
||||
"upstream_inference_cost": None,
|
||||
"upstream_inference_prompt_cost": 0,
|
||||
"upstream_inference_completions_cost": 0,
|
||||
},
|
||||
"completion_tokens_details": {
|
||||
"reasoning_tokens": 5,
|
||||
"image_tokens": 0,
|
||||
},
|
||||
},
|
||||
"cost": {
|
||||
"base_msats": 0,
|
||||
"input_msats": 0,
|
||||
"output_msats": 0,
|
||||
"total_msats": 0,
|
||||
},
|
||||
}
|
||||
return Response(json.dumps(content_dict).encode(), 200)
|
||||
|
||||
elif path.endswith("embeddings"):
|
||||
raise NotImplementedError
|
||||
elif path.endswith("responses"):
|
||||
raise NotImplementedError
|
||||
else:
|
||||
raise NotImplementedError
|
||||
|
||||
async def fetch_models(self) -> list[Model]:
|
||||
return [
|
||||
Model(
|
||||
id="mock/gpt-420-mock",
|
||||
name="mock/gpt-420-mock",
|
||||
created=0,
|
||||
description="mock model for testing",
|
||||
context_length=8192,
|
||||
architecture=Architecture(
|
||||
modality="text",
|
||||
input_modalities=["text"],
|
||||
output_modalities=["text"],
|
||||
tokenizer="",
|
||||
instruct_type=None,
|
||||
),
|
||||
pricing=Pricing(prompt=0.01, completion=0.01),
|
||||
),
|
||||
]
|
||||
|
||||
def transform_model_name(self, model_id: str) -> str:
|
||||
return "fake-model"
|
||||
|
||||
async def get_balance(self) -> float | None:
|
||||
return 420.69
|
||||
@@ -102,6 +102,8 @@ async def get_all_models_with_overrides(
|
||||
)
|
||||
for row in override_rows
|
||||
if row.upstream_provider_id is not None
|
||||
and row.upstream_provider_id in providers_by_id
|
||||
and providers_by_id[row.upstream_provider_id].enabled
|
||||
}
|
||||
|
||||
all_models: dict[str, Model] = {}
|
||||
@@ -218,14 +220,6 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
|
||||
results = await asyncio.gather(*tasks)
|
||||
upstreams = [p for p in results if p is not None]
|
||||
|
||||
if "https://testnut.cashu.space" in settings.cashu_mints:
|
||||
from .fake import MockUpstreamProvider
|
||||
|
||||
mock_provider = MockUpstreamProvider("mock", "mock")
|
||||
await mock_provider.refresh_models_cache()
|
||||
upstreams.append(mock_provider)
|
||||
logger.info("Initialized MockUpstreamProvider for testnut mint")
|
||||
|
||||
return upstreams
|
||||
|
||||
|
||||
|
||||
@@ -196,6 +196,37 @@ class PPQAIUpstreamProvider(BaseUpstreamProvider):
|
||||
)
|
||||
return []
|
||||
|
||||
async def on_upstream_error_redirect(
|
||||
self, status_code: int, error_message: str
|
||||
) -> None:
|
||||
if "insufficient balance" in error_message.lower():
|
||||
logger.warning(
|
||||
f"Disabling PPQ.AI provider ({self.base_url}) due to insufficient balance",
|
||||
extra={"error": error_message},
|
||||
)
|
||||
from sqlmodel import select
|
||||
|
||||
from ..core.db import UpstreamProviderRow, create_session
|
||||
|
||||
async with create_session() as session:
|
||||
statement = select(UpstreamProviderRow).where(
|
||||
UpstreamProviderRow.base_url == self.base_url
|
||||
and UpstreamProviderRow.api_key == self.api_key
|
||||
)
|
||||
result = await session.exec(statement)
|
||||
provider = result.first()
|
||||
|
||||
if provider:
|
||||
provider.enabled = False
|
||||
session.add(provider)
|
||||
await session.commit()
|
||||
|
||||
# Trigger re-initialization of providers
|
||||
# Import here to avoid circular dependency
|
||||
from ..proxy import reinitialize_upstreams
|
||||
|
||||
await reinitialize_upstreams()
|
||||
|
||||
async def create_account(self) -> dict[str, object]:
|
||||
"""Create a new PPQ.AI account.
|
||||
|
||||
|
||||
@@ -313,8 +313,6 @@ async def periodic_payout() -> None:
|
||||
try:
|
||||
async with db.create_session() as session:
|
||||
for mint_url in settings.cashu_mints:
|
||||
if mint_url == "https://testnut.cashu.space":
|
||||
continue
|
||||
for unit in ["sat", "msat"]:
|
||||
wallet = await get_wallet(mint_url, unit)
|
||||
proofs = get_proofs_per_mint_and_unit(
|
||||
|
||||
@@ -10,7 +10,6 @@ from httpx import AsyncClient
|
||||
|
||||
from .utils import (
|
||||
CashuTokenGenerator,
|
||||
PerformanceValidator,
|
||||
ResponseValidator,
|
||||
)
|
||||
|
||||
@@ -159,30 +158,7 @@ async def test_error_handling(
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_performance_requirements(integration_client: AsyncClient) -> None:
|
||||
"""Test that endpoints meet performance requirements"""
|
||||
|
||||
validator = PerformanceValidator()
|
||||
|
||||
# Test info endpoint performance
|
||||
for i in range(50):
|
||||
start = validator.start_timing("info_endpoint")
|
||||
response = await integration_client.get("/")
|
||||
validator.end_timing("info_endpoint", start)
|
||||
assert response.status_code == 200
|
||||
|
||||
# Validate 95th percentile is under 500ms
|
||||
result = validator.validate_response_time(
|
||||
"info_endpoint", max_duration=0.5, percentile=0.95
|
||||
)
|
||||
|
||||
assert result["valid"], (
|
||||
f"Performance requirement failed: "
|
||||
f"95th percentile was {result['percentile_time']:.3f}s "
|
||||
f"(required < {result['max_allowed']}s)"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
|
||||
@@ -9,6 +9,7 @@ import gc
|
||||
import statistics
|
||||
import time
|
||||
from typing import Any, Dict, List
|
||||
from unittest.mock import patch
|
||||
|
||||
import psutil
|
||||
import pytest
|
||||
@@ -105,42 +106,46 @@ class TestPerformanceBaseline:
|
||||
("GET", "/v1/wallet/info", authenticated_client, None),
|
||||
]
|
||||
|
||||
# Warm up
|
||||
for _ in range(10):
|
||||
await integration_client.get("/")
|
||||
# Enable provider discovery for this test
|
||||
with patch(
|
||||
"routstr.core.settings.settings.providers_refresh_interval_seconds", 300
|
||||
):
|
||||
# Warm up
|
||||
for _ in range(10):
|
||||
await integration_client.get("/")
|
||||
|
||||
# Test each endpoint
|
||||
for method, path, client, data in endpoints:
|
||||
response_times = []
|
||||
# Test each endpoint
|
||||
for method, path, client, data in endpoints:
|
||||
response_times = []
|
||||
|
||||
for i in range(100):
|
||||
start = time.time()
|
||||
for i in range(100):
|
||||
start = time.time()
|
||||
|
||||
if method == "GET":
|
||||
response = await client.get(path)
|
||||
else:
|
||||
response = await client.post(path, json=data)
|
||||
if method == "GET":
|
||||
response = await client.get(path)
|
||||
else:
|
||||
response = await client.post(path, json=data)
|
||||
|
||||
duration = time.time() - start
|
||||
response_times.append(duration * 1000) # Convert to ms
|
||||
duration = time.time() - start
|
||||
response_times.append(duration * 1000) # Convert to ms
|
||||
|
||||
assert response.status_code in [200, 201]
|
||||
assert response.status_code in [200, 201]
|
||||
|
||||
if i % 10 == 0:
|
||||
metrics.record_system_metrics()
|
||||
if i % 10 == 0:
|
||||
metrics.record_system_metrics()
|
||||
|
||||
# Verify 95th percentile < 500ms
|
||||
p95 = sorted(response_times)[int(len(response_times) * 0.95)]
|
||||
assert p95 < 500, (
|
||||
f"{method} {path} p95 response time {p95}ms exceeds 500ms limit"
|
||||
)
|
||||
# Verify 95th percentile < 500ms
|
||||
p95 = sorted(response_times)[int(len(response_times) * 0.95)]
|
||||
assert p95 < 500, (
|
||||
f"{method} {path} p95 response time {p95}ms exceeds 500ms limit"
|
||||
)
|
||||
|
||||
print(f"\n{method} {path}:")
|
||||
print(f" Mean: {statistics.mean(response_times):.2f}ms")
|
||||
print(f" P95: {p95:.2f}ms")
|
||||
print(
|
||||
f" P99: {sorted(response_times)[int(len(response_times) * 0.99)]:.2f}ms"
|
||||
)
|
||||
print(f"\n{method} {path}:")
|
||||
print(f" Mean: {statistics.mean(response_times):.2f}ms")
|
||||
print(f" P95: {p95:.2f}ms")
|
||||
print(
|
||||
f" P99: {sorted(response_times)[int(len(response_times) * 0.99)]:.2f}ms"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
|
||||
@@ -3,7 +3,7 @@ Integration tests for provider management functionality.
|
||||
Tests GET /v1/providers/ endpoint for listing and managing providers.
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
from typing import Any, Generator
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
@@ -11,7 +11,7 @@ from httpx import AsyncClient
|
||||
|
||||
from routstr.discovery import _PROVIDERS_CACHE
|
||||
|
||||
from .utils import PerformanceValidator, ResponseValidator
|
||||
from .utils import ResponseValidator
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
@@ -19,6 +19,15 @@ def _clear_providers_cache() -> None:
|
||||
_PROVIDERS_CACHE.clear()
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _enable_provider_discovery() -> Generator[None, Any, Any]:
|
||||
"""Enable provider discovery for all tests in this module"""
|
||||
with patch(
|
||||
"routstr.core.settings.settings.providers_refresh_interval_seconds", 300
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_providers_endpoint_default_response(
|
||||
@@ -518,46 +527,6 @@ async def test_providers_endpoint_response_format(
|
||||
assert isinstance(data_json["providers"], list)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_providers_endpoint_performance(integration_client: AsyncClient) -> None:
|
||||
"""Test providers endpoint meets performance requirements"""
|
||||
|
||||
# Mock quick responses to avoid network delays
|
||||
mock_events: list[dict[str, Any]] = [
|
||||
{
|
||||
"id": f"event{i}",
|
||||
"content": f"Provider: http://provider{i}.onion",
|
||||
"created_at": 1234567890 + i,
|
||||
}
|
||||
for i in range(5)
|
||||
]
|
||||
|
||||
validator = PerformanceValidator()
|
||||
|
||||
with patch(
|
||||
"routstr.discovery.query_nostr_relay_for_providers", return_value=mock_events
|
||||
):
|
||||
with patch("routstr.discovery.fetch_provider_health") as mock_fetch:
|
||||
mock_fetch.return_value = {"status_code": 200, "json": {"status": "online"}}
|
||||
|
||||
# Test multiple requests
|
||||
for i in range(10):
|
||||
start = validator.start_timing("providers_endpoint")
|
||||
response = await integration_client.get("/v1/providers/")
|
||||
validator.end_timing("providers_endpoint", start)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
# Validate performance (should be fast with mocked dependencies)
|
||||
perf_result = validator.validate_response_time(
|
||||
"providers_endpoint",
|
||||
max_duration=2.0, # Allow more time since it involves multiple operations
|
||||
percentile=0.95,
|
||||
)
|
||||
assert perf_result["valid"], f"Performance requirement failed: {perf_result}"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_providers_endpoint_concurrent_requests(
|
||||
|
||||
@@ -18,7 +18,6 @@ from routstr.core.db import ApiKey
|
||||
|
||||
from .utils import (
|
||||
ConcurrencyTester,
|
||||
PerformanceValidator,
|
||||
)
|
||||
|
||||
|
||||
@@ -551,39 +550,7 @@ async def test_proxy_get_concurrent_requests(
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_proxy_get_performance_requirements(
|
||||
integration_client: AsyncClient, authenticated_client: AsyncClient
|
||||
) -> None:
|
||||
"""Test that GET proxy requests meet performance requirements"""
|
||||
|
||||
validator = PerformanceValidator()
|
||||
|
||||
with patch("httpx.AsyncClient.request") as mock_request:
|
||||
mock_response = AsyncMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.headers = {"content-type": "application/json"}
|
||||
mock_response.json = MagicMock(return_value={"performance": "test"})
|
||||
mock_response.text = '{"performance": "test"}'
|
||||
mock_response.iter_bytes = AsyncMock(return_value=[b'{"performance": "test"}'])
|
||||
mock_request.return_value = mock_response
|
||||
|
||||
# Test multiple requests for performance measurement
|
||||
for i in range(20):
|
||||
start = validator.start_timing("proxy_get")
|
||||
response = await authenticated_client.get(f"/v1/perf-test-{i}")
|
||||
validator.end_timing("proxy_get", start)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
# Validate performance requirements
|
||||
perf_result = validator.validate_response_time(
|
||||
"proxy_get",
|
||||
max_duration=1.0, # Should complete within 1 second
|
||||
percentile=0.95,
|
||||
)
|
||||
assert perf_result["valid"], f"Performance requirement failed: {perf_result}"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
|
||||
@@ -15,7 +15,6 @@ from httpx import ASGITransport, AsyncClient
|
||||
|
||||
from .utils import (
|
||||
ConcurrencyTester,
|
||||
PerformanceValidator,
|
||||
)
|
||||
|
||||
|
||||
@@ -290,55 +289,7 @@ async def test_proxy_post_unauthorized_access(integration_client: AsyncClient) -
|
||||
assert response.status_code in [400, 401]
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
async def test_proxy_post_performance(
|
||||
integration_client: AsyncClient, authenticated_client: AsyncClient
|
||||
) -> None:
|
||||
"""Test POST endpoint performance requirements"""
|
||||
|
||||
test_payload = {
|
||||
"model": "gpt-3.5-turbo",
|
||||
"messages": [{"role": "user", "content": "Performance test"}],
|
||||
}
|
||||
|
||||
validator = PerformanceValidator()
|
||||
|
||||
with patch("httpx.AsyncClient.send") as mock_send:
|
||||
# Mock fast responses
|
||||
async def mock_iter_bytes(*args: Any, **kwargs: Any) -> Any:
|
||||
yield b'{"choices": [{"message": {"content": "Fast"}}], "usage": {"total_tokens": 5}}'
|
||||
|
||||
mock_response = AsyncMock()
|
||||
mock_response.status_code = 200
|
||||
mock_response.headers = {"content-type": "application/json"}
|
||||
response_data = {
|
||||
"choices": [{"message": {"content": "Fast"}}],
|
||||
"usage": {"total_tokens": 5},
|
||||
}
|
||||
mock_response.text = json.dumps(response_data)
|
||||
mock_response.json = AsyncMock(return_value=response_data)
|
||||
mock_response.iter_bytes = mock_iter_bytes
|
||||
mock_response.aiter_bytes = mock_iter_bytes
|
||||
mock_send.return_value = mock_response
|
||||
|
||||
# Run multiple requests for performance measurement
|
||||
for i in range(20):
|
||||
start = validator.start_timing("proxy_post")
|
||||
response = await authenticated_client.post(
|
||||
"/v1/chat/completions", json=test_payload
|
||||
)
|
||||
validator.end_timing("proxy_post", start)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
# Validate performance
|
||||
perf_result = validator.validate_response_time(
|
||||
"proxy_post",
|
||||
max_duration=1.5, # Allow slightly more time for POST
|
||||
percentile=0.95,
|
||||
)
|
||||
assert perf_result["valid"], f"Performance requirement failed: {perf_result}"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
|
||||
@@ -3,7 +3,7 @@ Integration tests for wallet information retrieval endpoints.
|
||||
Tests GET /v1/wallet/ and GET /v1/wallet/info endpoints with various scenarios.
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Any
|
||||
|
||||
@@ -367,30 +367,4 @@ async def test_wallet_info_with_special_characters_in_headers(
|
||||
# Note: Current implementation doesn't return refund_address in response
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.slow
|
||||
async def test_wallet_endpoints_performance(authenticated_client: AsyncClient) -> None:
|
||||
"""Test wallet endpoints meet performance requirements"""
|
||||
|
||||
# Warm up
|
||||
await authenticated_client.get("/v1/wallet/")
|
||||
|
||||
# Measure response times
|
||||
response_times = []
|
||||
|
||||
for _ in range(50):
|
||||
start_time = time.time()
|
||||
response = await authenticated_client.get("/v1/wallet/")
|
||||
end_time = time.time()
|
||||
|
||||
assert response.status_code == 200
|
||||
response_times.append(end_time - start_time)
|
||||
|
||||
# Calculate statistics
|
||||
avg_time = sum(response_times) / len(response_times)
|
||||
max_time = max(response_times)
|
||||
|
||||
# Performance assertions
|
||||
assert avg_time < 0.1 # Average should be under 100ms
|
||||
assert max_time < 0.5 # No request should take more than 500ms
|
||||
|
||||
@@ -537,42 +537,4 @@ async def test_refund_with_expired_key(
|
||||
assert response.json()["recipient"] == "expired@ln.address"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.slow
|
||||
async def test_refund_performance(
|
||||
integration_client: AsyncClient, testmint_wallet: Any
|
||||
) -> None:
|
||||
"""Test refund endpoint performance"""
|
||||
|
||||
import time
|
||||
|
||||
# Create multiple API keys
|
||||
api_keys = []
|
||||
for i in range(10):
|
||||
token = await testmint_wallet.mint_tokens(100 + i)
|
||||
# Use cashu token as Bearer auth to create API key
|
||||
integration_client.headers["Authorization"] = f"Bearer {token}"
|
||||
response = await integration_client.get("/v1/wallet/info")
|
||||
assert response.status_code == 200
|
||||
api_keys.append(response.json()["api_key"])
|
||||
|
||||
# Measure refund times
|
||||
refund_times = []
|
||||
|
||||
for api_key in api_keys:
|
||||
integration_client.headers["Authorization"] = f"Bearer {api_key}"
|
||||
|
||||
start_time = time.time()
|
||||
response = await integration_client.post("/v1/wallet/refund")
|
||||
end_time = time.time()
|
||||
|
||||
assert response.status_code == 200
|
||||
refund_times.append(end_time - start_time)
|
||||
|
||||
# Performance assertions
|
||||
avg_time = sum(refund_times) / len(refund_times)
|
||||
max_time = max(refund_times)
|
||||
|
||||
assert avg_time < 0.5 # Average under 500ms
|
||||
assert max_time < 1.0 # No refund takes more than 1 second
|
||||
|
||||
@@ -10,7 +10,6 @@ os.environ["UPSTREAM_API_KEY"] = "test"
|
||||
from routstr.algorithm import ( # noqa: E402
|
||||
calculate_model_cost_score,
|
||||
get_provider_penalty,
|
||||
should_prefer_model,
|
||||
)
|
||||
from routstr.payment.models import Architecture, Model, Pricing # noqa: E402
|
||||
|
||||
@@ -100,100 +99,3 @@ def test_get_provider_penalty_openrouter() -> None:
|
||||
provider = create_test_provider("openrouter", "https://openrouter.ai/api/v1")
|
||||
penalty = get_provider_penalty(provider)
|
||||
assert penalty == 1.001
|
||||
|
||||
|
||||
def test_should_prefer_model_cheaper_wins() -> None:
|
||||
"""Test that cheaper model is preferred."""
|
||||
cheap_model = create_test_model("cheap", prompt_price=0.001, completion_price=0.002)
|
||||
expensive_model = create_test_model(
|
||||
"expensive", prompt_price=0.03, completion_price=0.06
|
||||
)
|
||||
|
||||
provider1 = create_test_provider("provider1")
|
||||
provider2 = create_test_provider("provider2")
|
||||
|
||||
# Cheaper model should win
|
||||
assert should_prefer_model(
|
||||
cheap_model, provider1, expensive_model, provider2, "test-alias"
|
||||
)
|
||||
|
||||
# More expensive model should not win
|
||||
assert not should_prefer_model(
|
||||
expensive_model, provider2, cheap_model, provider1, "test-alias"
|
||||
)
|
||||
|
||||
|
||||
def test_should_prefer_model_exact_match_wins() -> None:
|
||||
"""Test that exact alias match beats cheaper price."""
|
||||
# Make model IDs match the alias differently
|
||||
exact_match = create_test_model(
|
||||
"test-model", prompt_price=0.03, completion_price=0.06
|
||||
)
|
||||
no_match = create_test_model(
|
||||
"other-model", prompt_price=0.001, completion_price=0.002
|
||||
)
|
||||
|
||||
provider1 = create_test_provider("provider1")
|
||||
provider2 = create_test_provider("provider2")
|
||||
|
||||
# Exact match should win even though it's more expensive
|
||||
assert should_prefer_model(
|
||||
exact_match, provider1, no_match, provider2, "test-model"
|
||||
)
|
||||
|
||||
|
||||
def test_should_prefer_model_openrouter_slight_penalty() -> None:
|
||||
"""Test that OpenRouter has slight penalty compared to other providers."""
|
||||
model1 = create_test_model("model1", prompt_price=0.001, completion_price=0.002)
|
||||
model2 = create_test_model("model2", prompt_price=0.001, completion_price=0.002)
|
||||
|
||||
regular_provider = create_test_provider("regular", "http://provider.com")
|
||||
openrouter_provider = create_test_provider(
|
||||
"openrouter", "https://openrouter.ai/api/v1"
|
||||
)
|
||||
|
||||
# Regular provider should be preferred over OpenRouter at same cost
|
||||
assert should_prefer_model(
|
||||
model1, regular_provider, model2, openrouter_provider, "test-alias"
|
||||
)
|
||||
|
||||
# OpenRouter should not replace regular provider at same cost
|
||||
assert not should_prefer_model(
|
||||
model2, openrouter_provider, model1, regular_provider, "test-alias"
|
||||
)
|
||||
|
||||
|
||||
def test_should_prefer_model_openrouter_can_win_if_cheaper() -> None:
|
||||
"""Test that OpenRouter can still win if significantly cheaper."""
|
||||
cheap_model = create_test_model(
|
||||
"cheap", prompt_price=0.0001, completion_price=0.0002
|
||||
)
|
||||
expensive_model = create_test_model(
|
||||
"expensive", prompt_price=0.03, completion_price=0.06
|
||||
)
|
||||
|
||||
regular_provider = create_test_provider("regular", "http://provider.com")
|
||||
openrouter_provider = create_test_provider(
|
||||
"openrouter", "https://openrouter.ai/api/v1"
|
||||
)
|
||||
|
||||
# OpenRouter should win if it's much cheaper (even with penalty)
|
||||
assert should_prefer_model(
|
||||
cheap_model,
|
||||
openrouter_provider,
|
||||
expensive_model,
|
||||
regular_provider,
|
||||
"test-alias",
|
||||
)
|
||||
|
||||
|
||||
def test_should_prefer_model_same_cost_first_wins() -> None:
|
||||
"""Test that when costs are identical, current model is kept."""
|
||||
model1 = create_test_model("model1", prompt_price=0.001, completion_price=0.002)
|
||||
model2 = create_test_model("model2", prompt_price=0.001, completion_price=0.002)
|
||||
|
||||
provider1 = create_test_provider("provider1")
|
||||
provider2 = create_test_provider("provider2")
|
||||
|
||||
# When costs are equal, should not replace
|
||||
assert not should_prefer_model(model2, provider2, model1, provider1, "test-alias")
|
||||
|
||||
Reference in New Issue
Block a user