From 100654adb8a3fa2cf4d09ff6c85fc006806d9844 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Mon, 20 Apr 2026 17:45:22 +0800 Subject: [PATCH 1/2] feat: Introduce random selection strategy and cooldown configuration for account management --- app/dataplane/account/__init__.py | 44 ++++- app/dataplane/account/feedback.py | 94 ++++++--- app/dataplane/account/selector.py | 301 ++++++++++++++++++++--------- app/dataplane/account/table.py | 6 + app/dataplane/shared/enums.py | 3 + app/main.py | 25 ++- app/products/_account_selection.py | 25 ++- app/products/anthropic/messages.py | 4 +- app/products/openai/chat.py | 4 +- app/products/openai/responses.py | 4 +- app/products/web/admin/__init__.py | 2 + app/statics/admin/account.html | 34 +++- app/statics/admin/config.html | 167 ++++++++++++++++ app/statics/i18n/de.json | 17 +- app/statics/i18n/en.json | 23 +++ app/statics/i18n/es.json | 17 +- app/statics/i18n/fr.json | 17 +- app/statics/i18n/ja.json | 17 +- app/statics/i18n/zh.json | 23 +++ config.defaults.toml | 14 ++ 20 files changed, 687 insertions(+), 154 deletions(-) diff --git a/app/dataplane/account/__init__.py b/app/dataplane/account/__init__.py index b753edcb3..20937f366 100644 --- a/app/dataplane/account/__init__.py +++ b/app/dataplane/account/__init__.py @@ -8,16 +8,17 @@ import asyncio from typing import TYPE_CHECKING +from app.platform.config.snapshot import get_config from app.platform.logging.logger import logger from app.platform.runtime.clock import now_s from app.control.account.repository import AccountRepository from app.control.account.enums import FeedbackKind from .table import AccountRuntimeTable from .lease import AccountLease, new_lease -from .selector import select, select_any +from .selector import current_strategy, select, select_any from .sync import bootstrap as _bootstrap, apply_changes from . import feedback as fb -from ..shared.enums import StatusId +from ..shared.enums import POOL_ID_TO_STR, StatusId if TYPE_CHECKING: pass @@ -250,12 +251,22 @@ async def feedback( ts = now_s_val if now_s_val is not None else now_s() + strategy = current_strategy() + async with self._lock: if kind == FeedbackKind.SUCCESS: - fb.apply_success(table, idx, mode_id) + if strategy == "random": + fb.apply_success_random(table, idx) + else: + fb.apply_success_quota(table, idx, mode_id) elif kind == FeedbackKind.RATE_LIMITED: - fb.apply_rate_limited(table, idx, mode_id) + if strategy == "random": + pool_id = int(table.pool_by_idx[idx]) + cooling_sec = _pool_cooling_sec(pool_id) + fb.apply_rate_limited_random(table, idx, cooling_sec=cooling_sec) + else: + fb.apply_rate_limited_quota(table, idx, mode_id) fb.update_last_fail(table, idx, ts) elif kind == FeedbackKind.UNAUTHORIZED: @@ -271,8 +282,9 @@ async def feedback( fb.apply_server_error(table, idx) fb.update_last_fail(table, idx, ts) - # Apply authoritative quota data from upstream response headers. - if remaining is not None and reset_at_ms is not None: + # Quota strategy may receive authoritative quota data from upstream + # response headers; the random strategy ignores this entirely. + if strategy == "quota" and remaining is not None and reset_at_ms is not None: reset_s = int(reset_at_ms // 1000) fb.apply_quota_update(table, idx, mode_id, remaining, reset_s) @@ -289,6 +301,26 @@ def revision(self) -> int: return self._table.revision if self._table else 0 +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +_DEFAULT_COOLING_HOURS: dict[str, int] = { + "basic": 20, + "super": 2, + "heavy": 2, +} + + +def _pool_cooling_sec(pool_id: int) -> int: + """Cooling seconds for a 429 on a given pool (random strategy only).""" + pool_str = POOL_ID_TO_STR.get(pool_id, "basic") + default_hours = _DEFAULT_COOLING_HOURS.get(pool_str, 2) + hours = int(get_config(f"account.random.cooling_{pool_str}_hours", default_hours)) + return max(0, hours) * 3600 + + # --------------------------------------------------------------------------- # Module-level singleton # --------------------------------------------------------------------------- diff --git a/app/dataplane/account/feedback.py b/app/dataplane/account/feedback.py index b77942c1e..3372b56e3 100644 --- a/app/dataplane/account/feedback.py +++ b/app/dataplane/account/feedback.py @@ -1,39 +1,78 @@ """Apply feedback to the runtime table columns (in-place, lock-free inner ops). -The caller (AccountDirectory) is responsible for holding the state lock -before calling these functions. +The caller (AccountDirectory) is responsible for holding the state lock before +calling these functions. + +Strategy-split functions +------------------------ +``apply_success`` / ``apply_rate_limited`` are split into ``*_quota`` and +``*_random`` variants. The caller picks exactly one based on the active +selector strategy. No runtime ``if`` inside the apply helpers themselves. """ +from app.platform.runtime.clock import now_s from ..shared.enums import ALL_MODE_IDS, StatusId from .table import AccountRuntimeTable # Health adjustment constants. -_SUCCESS_STEP = 0.12 -_AUTH_FACTOR = 0.55 -_FORBIDDEN_FACTOR = 0.25 -_RATE_LIMIT_FACTOR = 0.45 +_SUCCESS_STEP = 0.12 +_AUTH_FACTOR = 0.55 +_FORBIDDEN_FACTOR = 0.25 +_RATE_LIMIT_FACTOR = 0.45 _SERVER_ERROR_FACTOR = 0.75 _MIN_HEALTH = 0.05 _MAX_HEALTH = 1.0 -def apply_success(table: AccountRuntimeTable, idx: int, mode_id: int) -> None: - """Record a successful call: decrement quota, improve health.""" +# --------------------------------------------------------------------------- +# apply_success — two independent implementations +# --------------------------------------------------------------------------- + + +def apply_success_quota(table: AccountRuntimeTable, idx: int, mode_id: int) -> None: + """Quota strategy: decrement per-mode quota and improve health.""" quota_col = table._quota_col(mode_id) - new_q = max(0, int(quota_col[idx]) - 1) - quota_col[idx] = new_q + quota_col[idx] = max(0, int(quota_col[idx]) - 1) + _bump_health(table, idx) + - # Recover health. - h = min(_MAX_HEALTH, float(table.health_by_idx[idx]) + _SUCCESS_STEP) - table.health_by_idx[idx] = h +def apply_success_random(table: AccountRuntimeTable, idx: int) -> None: + """Random strategy: only improve health; quota columns are ignored.""" + _bump_health(table, idx) -def apply_rate_limited(table: AccountRuntimeTable, idx: int, mode_id: int) -> None: - """Zero the mode quota and reduce health.""" +# --------------------------------------------------------------------------- +# apply_rate_limited — two independent implementations +# --------------------------------------------------------------------------- + + +def apply_rate_limited_quota( + table: AccountRuntimeTable, idx: int, mode_id: int +) -> None: + """Quota strategy: zero the mode quota and reduce health.""" table._quota_col(mode_id)[idx] = 0 _adjust_health(table, idx, _RATE_LIMIT_FACTOR) +def apply_rate_limited_random( + table: AccountRuntimeTable, idx: int, *, cooling_sec: int +) -> None: + """Random strategy: set per-account cooldown timestamp; do not touch quota. + + The account stays in ``mode_available``; ``_random_select`` filters it out + until ``cooling_until_s_by_idx`` has elapsed. + """ + ts = now_s() + max(0, cooling_sec) + cooling_col = table.cooling_until_s_by_idx + cooling_col[idx] = max(int(cooling_col[idx]), ts) + _adjust_health(table, idx, _RATE_LIMIT_FACTOR) + + +# --------------------------------------------------------------------------- +# Strategy-agnostic feedback helpers (shared) +# --------------------------------------------------------------------------- + + def apply_auth_failure(table: AccountRuntimeTable, idx: int) -> None: """Reduce health on 401; caller may mark account expired.""" _adjust_health(table, idx, _AUTH_FACTOR) @@ -62,13 +101,11 @@ def apply_status_change( table.status_by_idx[idx] = new_status_id if new_status_id != int(StatusId.ACTIVE): - # Remove from all mode availability buckets. for mode_id in ALL_MODE_IDS: bucket = table.mode_available.get((pool_id, mode_id)) if bucket: bucket.discard(idx) else: - # Re-add to mode buckets where quota > 0. for mode_id in ALL_MODE_IDS: if int(table._quota_col(mode_id)[idx]) > 0: table.mode_available.setdefault((pool_id, mode_id), set()).add(idx) @@ -81,7 +118,11 @@ def apply_quota_update( remaining: int, reset_s: int, ) -> None: - """Update quota and reset timestamp from upstream API data.""" + """Update quota and reset timestamp from upstream API data. + + Used exclusively by the quota strategy (refresh service). The random + strategy never calls this. + """ quota_col = table._quota_col(mode_id) reset_col = table._reset_col(mode_id) quota_col[idx] = max(0, min(remaining, 32767)) @@ -116,14 +157,23 @@ def update_last_fail(table: AccountRuntimeTable, idx: int, now_s: int) -> None: # --------------------------------------------------------------------------- +def _bump_health(table: AccountRuntimeTable, idx: int) -> None: + table.health_by_idx[idx] = min( + _MAX_HEALTH, float(table.health_by_idx[idx]) + _SUCCESS_STEP + ) + + def _adjust_health(table: AccountRuntimeTable, idx: int, factor: float) -> None: - h = max(_MIN_HEALTH, float(table.health_by_idx[idx]) * factor) - table.health_by_idx[idx] = h + table.health_by_idx[idx] = max( + _MIN_HEALTH, float(table.health_by_idx[idx]) * factor + ) __all__ = [ - "apply_success", - "apply_rate_limited", + "apply_success_quota", + "apply_success_random", + "apply_rate_limited_quota", + "apply_rate_limited_random", "apply_auth_failure", "apply_forbidden", "apply_server_error", diff --git a/app/dataplane/account/selector.py b/app/dataplane/account/selector.py index 814259049..c7d048cbe 100644 --- a/app/dataplane/account/selector.py +++ b/app/dataplane/account/selector.py @@ -1,22 +1,63 @@ -"""Hot-path account selector — pure function, no object allocation. +"""Hot-path account selector — pluggable strategies. -All scoring operates on typed ``array.array`` columns; no attribute lookup -overhead. The caller provides the frozen set of excluded indices and -pre-resolved tag index set. +Two fully independent strategies: + +* ``_quota_select`` — scores candidates by health / quota / inflight / fails. + Used when ``account.refresh.enabled=true``. Behaviour is the historical one, + kept unchanged. +* ``_random_select`` — uniform random choice among non-cooling candidates. + Used when ``account.refresh.enabled=false``. Ignores quota and health entirely. + +Strategy selection is process-global, registered once by the lifespan via +:func:`set_strategy`. Callers (``AccountDirectory``) invoke :func:`select` / +:func:`select_any` and are unaware of which strategy is active. """ +import array # noqa: F401 — used in forward-referenced type annotations +import random +from typing import Literal + +from app.platform.config.snapshot import get_config from ..shared.enums import PoolId from .table import AccountRuntimeTable -# Scoring weights — tuned for throughput/fairness balance. -_W_HEALTH = 100.0 -_W_QUOTA = 25.0 -_W_RECENT = 15.0 # penalty for recently used accounts +# Scoring weights used by the quota strategy. +_W_HEALTH = 100.0 +_W_QUOTA = 25.0 +_W_RECENT = 15.0 # penalty for recently used accounts _W_INFLIGHT = 20.0 -_W_FAIL = 4.0 +_W_FAIL = 4.0 _RECENT_WINDOW_S = 15 # seconds +# --------------------------------------------------------------------------- +# Strategy registry +# --------------------------------------------------------------------------- + +_StrategyName = Literal["quota", "random"] +_STRATEGY_NAME: _StrategyName = "random" + + +def set_strategy(name: _StrategyName) -> None: + """Register the process-global selection strategy. + + Called once by the lifespan after reading ``account.refresh.enabled``. + """ + global _STRATEGY_NAME + if name not in ("quota", "random"): + raise ValueError(f"unknown selection strategy: {name!r}") + _STRATEGY_NAME = name + + +def current_strategy() -> _StrategyName: + return _STRATEGY_NAME + + +# --------------------------------------------------------------------------- +# Public entry points — delegate to the active strategy +# --------------------------------------------------------------------------- + + def select( table: AccountRuntimeTable, pool_id: int, @@ -26,30 +67,79 @@ def select( prefer_tag_idxs: set[int] | None = None, now_s: int, ) -> int | None: - """Select the best available account slot index. + """Select an account slot for ``(pool_id, mode_id)``. + + Returns the slot index or ``None`` when no candidate is available. + Does not mutate the table — callers increment inflight separately. + """ + if _STRATEGY_NAME == "random": + return _random_select( + table, pool_id, + exclude_idxs=exclude_idxs, + prefer_tag_idxs=prefer_tag_idxs, + now_s=now_s, + ) + return _quota_select( + table, pool_id, mode_id, + exclude_idxs=exclude_idxs, + prefer_tag_idxs=prefer_tag_idxs, + now_s=now_s, + ) + + +def select_any( + table: AccountRuntimeTable, + pool_id: int, + *, + exclude_idxs: frozenset[int] | None = None, + prefer_tag_idxs: set[int] | None = None, + now_s: int, +) -> int | None: + """Select any active account in ``pool_id`` irrespective of per-mode quota. - Returns the slot index on success, ``None`` if no account is available. - Does **not** mutate any table state — callers must increment inflight. + Used by WebSocket-based products that manage their own rate limiting. """ + if _STRATEGY_NAME == "random": + return _random_select( + table, pool_id, + exclude_idxs=exclude_idxs, + prefer_tag_idxs=prefer_tag_idxs, + now_s=now_s, + ) + return _quota_select_any( + table, pool_id, + exclude_idxs=exclude_idxs, + prefer_tag_idxs=prefer_tag_idxs, + now_s=now_s, + ) + + +# --------------------------------------------------------------------------- +# Strategy: quota — score-based selection (unchanged behaviour) +# --------------------------------------------------------------------------- + + +def _quota_select( + table: AccountRuntimeTable, + pool_id: int, + mode_id: int, + *, + exclude_idxs: frozenset[int] | None, + prefer_tag_idxs: set[int] | None, + now_s: int, +) -> int | None: candidates: set[int] | None = table.mode_available.get((pool_id, mode_id)) if not candidates: return None - # Apply window-expiry resets for basic accounts inline. - reset_col = table._reset_col(mode_id) - quota_col = table._quota_col(mode_id) - total_col = table._total_col(mode_id) + reset_col = table._reset_col(mode_id) + quota_col = table._quota_col(mode_id) + total_col = table._total_col(mode_id) window_col = table._window_col(mode_id) _maybe_reset_windows( - table, - candidates, - mode_id, - reset_col, - quota_col, - total_col, - window_col, - pool_id, - now_s, + table, candidates, mode_id, + reset_col, quota_col, total_col, window_col, + pool_id, now_s, ) working: set[int] = candidates.copy() @@ -59,30 +149,50 @@ def select( if not working: return None - # Prefer tagged subset; fall back to full set if empty. if prefer_tag_idxs: preferred = working & prefer_tag_idxs working = preferred if preferred else working - return _best(table, working, mode_id, quota_col, now_s) + return _best(table, working, quota_col, now_s) + + +def _quota_select_any( + table: AccountRuntimeTable, + pool_id: int, + *, + exclude_idxs: frozenset[int] | None, + prefer_tag_idxs: set[int] | None, + now_s: int, +) -> int | None: + candidates: set[int] = _pool_union(table, pool_id) + if not candidates: + return None + + working = candidates.copy() + if exclude_idxs: + working -= exclude_idxs + if not working: + return None + + if prefer_tag_idxs: + preferred = working & prefer_tag_idxs + working = preferred if preferred else working + + return _best_no_quota(table, working, now_s) def _maybe_reset_windows( - table: AccountRuntimeTable, + table: AccountRuntimeTable, candidates: set[int], - mode_id: int, + mode_id: int, reset_col: "array.array", quota_col: "array.array", total_col: "array.array", window_col: "array.array", - pool_id: int, - now_s: int, + pool_id: int, + now_s: int, ) -> None: - """Reset expired windows for basic-pool accounts (no API call required). - - Only applies to the basic pool — super/heavy quotas are managed exclusively - by the periodic refresh service and must not be reset inline. - """ + """Reset expired windows for basic-pool accounts inline (no API call needed).""" if pool_id != int(PoolId.BASIC): return @@ -93,22 +203,19 @@ def _maybe_reset_windows( if int(table.pool_by_idx[idx]) != pool_id: continue new_total = int(total_col[idx]) - window_s = int(window_col[idx]) + window_s = int(window_col[idx]) if new_total <= 0 or window_s <= 0: continue - # Window expired — restore the last known total for this account/mode. quota_col[idx] = new_total reset_col[idx] = now_s + window_s def _best( - table: AccountRuntimeTable, - working: set[int], - mode_id: int, + table: AccountRuntimeTable, + working: set[int], quota_col: "array.array", - now_s: int, -) -> int: - """Return the index of the highest-scoring candidate.""" + now_s: int, +) -> int | None: best_idx = -1 best_score = -1e18 @@ -118,7 +225,7 @@ def _best( last_use_col = table.last_use_at_by_idx for idx in working: - quota = int(quota_col[idx]) + quota = int(quota_col[idx]) if quota <= 0: continue health = float(health_col[idx]) @@ -144,60 +251,23 @@ def _best( return best_idx if best_idx >= 0 else None -def select_any( - table: AccountRuntimeTable, - pool_id: int, - *, - exclude_idxs: frozenset[int] | None = None, - prefer_tag_idxs: set[int] | None = None, - now_s: int, -) -> int | None: - """Select any active account from *pool_id* without mode-specific quota checking. - - Used for WebSocket-based operations that manage their own upstream rate limiting - and do not consume tracked quota buckets. Candidates are the union of all - mode_available sets for the pool so that accounts are reachable even when one - particular mode bucket (e.g. AUTO) is exhausted. - """ - candidates: set[int] = set() - for (pid, _mid), accounts in table.mode_available.items(): - if pid == pool_id: - candidates |= accounts - - if not candidates: - return None - - working = candidates.copy() - if exclude_idxs: - working -= exclude_idxs - if not working: - return None - - if prefer_tag_idxs: - preferred = working & prefer_tag_idxs - working = preferred if preferred else working - - return _best_no_quota(table, working, now_s) - - def _best_no_quota( table: AccountRuntimeTable, working: set[int], now_s: int, ) -> int | None: - """Score candidates by health and inflight only — no quota weighting.""" - best_idx = -1 + best_idx = -1 best_score = -1e18 - health_col = table.health_by_idx + health_col = table.health_by_idx inflight_col = table.inflight_by_idx - fail_col = table.fail_count_by_idx + fail_col = table.fail_count_by_idx last_use_col = table.last_use_at_by_idx for idx in working: - health = float(health_col[idx]) + health = float(health_col[idx]) inflight = int(inflight_col[idx]) - fails = min(int(fail_col[idx]), 10) + fails = min(int(fail_col[idx]), 10) last_use = int(last_use_col[idx]) score = health * _W_HEALTH - inflight * _W_INFLIGHT - fails * _W_FAIL @@ -208,9 +278,62 @@ def _best_no_quota( if score > best_score: best_score = score - best_idx = idx + best_idx = idx return best_idx if best_idx >= 0 else None -__all__ = ["select", "select_any"] +# --------------------------------------------------------------------------- +# Strategy: random — uniform choice with cooling + inflight filter +# --------------------------------------------------------------------------- + + +def _random_select( + table: AccountRuntimeTable, + pool_id: int, + *, + exclude_idxs: frozenset[int] | None, + prefer_tag_idxs: set[int] | None, + now_s: int, +) -> int | None: + candidates: set[int] = _pool_union(table, pool_id) + if not candidates: + return None + + max_inflight = int(get_config("account.selection.max_inflight", 8)) + cooling_col = table.cooling_until_s_by_idx + inflight_col = table.inflight_by_idx + + working = candidates.copy() + if exclude_idxs: + working -= exclude_idxs + working = { + idx for idx in working + if int(cooling_col[idx]) <= now_s + and int(inflight_col[idx]) < max_inflight + } + if not working: + return None + + if prefer_tag_idxs: + preferred = working & prefer_tag_idxs + working = preferred if preferred else working + + return random.choice(tuple(working)) + + +# --------------------------------------------------------------------------- +# Shared helper +# --------------------------------------------------------------------------- + + +def _pool_union(table: AccountRuntimeTable, pool_id: int) -> set[int]: + """Union of all ``mode_available`` buckets for ``pool_id``.""" + out: set[int] = set() + for (pid, _mid), accounts in table.mode_available.items(): + if pid == pool_id: + out |= accounts + return out + + +__all__ = ["select", "select_any", "set_strategy", "current_strategy"] diff --git a/app/dataplane/account/table.py b/app/dataplane/account/table.py index 69ab143a8..16a5719e9 100644 --- a/app/dataplane/account/table.py +++ b/app/dataplane/account/table.py @@ -151,6 +151,11 @@ class AccountRuntimeTable: default_factory=lambda: array.array("L") ) + # --- Per-account cooldown (random strategy only; uint32 epoch-seconds; 0 = not cooling) --- + cooling_until_s_by_idx: "array.array[int]" = field( + default_factory=lambda: array.array("L") + ) + # --- Pre-computed selection indexes --- # (pool_id, mode_id) → set of idx with a supported quota window and status == ACTIVE mode_available: dict[tuple[int, int], set[int]] = field(default_factory=dict) @@ -300,6 +305,7 @@ def _append_slot( self.health_by_idx.append(health) self.last_use_at_by_idx.append(last_use_s) self.last_fail_at_by_idx.append(last_fail_s) + self.cooling_until_s_by_idx.append(0) self.size += 1 self._add_to_indexes(idx) self._add_to_tag_idx(idx, tags) diff --git a/app/dataplane/shared/enums.py b/app/dataplane/shared/enums.py index bdbe4a9ea..88b9ed66b 100644 --- a/app/dataplane/shared/enums.py +++ b/app/dataplane/shared/enums.py @@ -36,6 +36,8 @@ class StatusId(IntEnum): "heavy": int(PoolId.HEAVY), } +POOL_ID_TO_STR: dict[int, str] = {v: k for k, v in POOL_STR_TO_ID.items()} + STATUS_STR_TO_ID: dict[str, int] = { "active": int(StatusId.ACTIVE), "cooling": int(StatusId.COOLING), @@ -56,6 +58,7 @@ class StatusId(IntEnum): "PoolId", "StatusId", "POOL_STR_TO_ID", + "POOL_ID_TO_STR", "STATUS_STR_TO_ID", "ALL_MODE_IDS", ] diff --git a/app/main.py b/app/main.py index 9a5174928..60bb560a2 100644 --- a/app/main.py +++ b/app/main.py @@ -182,7 +182,16 @@ async def _sync_loop() -> None: # 4. Account refresh scheduler — only the leader worker. # Uses an advisory file lock so exactly one process runs the heavy # upstream quota-fetch loop regardless of worker count. + # + # ``account.refresh.enabled`` selects between two independent runtime + # strategies: + # - true → "quota" selector + scheduler running (default). + # - false → "random" selector + scheduler idle (no upstream probing). from app.control.account.refresh import AccountRefreshService + from app.dataplane.account.selector import set_strategy as _set_selection_strategy + + refresh_enabled = _config.get_bool("account.refresh.enabled", False) + _set_selection_strategy("quota" if refresh_enabled else "random") refresh_svc = AccountRefreshService(repo) set_refresh_service(refresh_svc) @@ -190,18 +199,28 @@ async def _sync_loop() -> None: is_leader = _try_acquire_scheduler_lock() scheduler = get_account_refresh_scheduler(refresh_svc) - if is_leader: + strategy_name = "quota" if refresh_enabled else "random" + if is_leader and refresh_enabled: scheduler.start() logger.info( - "scheduler leader: pid={} active_sync_s={} idle_sync_s={}", + "scheduler leader: pid={} strategy=quota active_sync_s={} idle_sync_s={}", + os.getpid(), + _SYNC_ACTIVE_INTERVAL, + _SYNC_IDLE_INTERVAL, + ) + elif is_leader: + logger.info( + "scheduler leader: pid={} strategy=random (scheduler idle) " + "active_sync_s={} idle_sync_s={}", os.getpid(), _SYNC_ACTIVE_INTERVAL, _SYNC_IDLE_INTERVAL, ) else: logger.info( - "scheduler follower: pid={} active_sync_s={} idle_sync_s={}", + "scheduler follower: pid={} strategy={} active_sync_s={} idle_sync_s={}", os.getpid(), + strategy_name, _SYNC_ACTIVE_INTERVAL, _SYNC_IDLE_INTERVAL, ) diff --git a/app/products/_account_selection.py b/app/products/_account_selection.py index 056583e67..a2f9423d9 100644 --- a/app/products/_account_selection.py +++ b/app/products/_account_selection.py @@ -3,8 +3,26 @@ from app.control.model.enums import ModeId from app.control.model.spec import ModelSpec from app.control.account.runtime import get_refresh_service +from app.dataplane.account.selector import current_strategy from app.platform.config.snapshot import get_config +# Random strategy has no config key for retry count; it is pinned here so that +# every retry-driven call site (chat / images / video / anthropic) sees the same +# value without introducing scattered magic numbers. +_RANDOM_MAX_RETRIES = 5 + + +def selection_max_retries() -> int: + """Retry count for account-swap loops, aware of the active selection strategy. + + - ``random`` strategy: fixed at :data:`_RANDOM_MAX_RETRIES` (=5). + - ``quota`` strategy: reads ``retry.max_retries`` (default 1), preserving + the historical behaviour. + """ + if current_strategy() == "random": + return _RANDOM_MAX_RETRIES + return int(get_config("retry.max_retries", 1)) + def mode_candidates(spec: ModelSpec) -> tuple[int, ...]: """Return mode IDs to try for *spec* in priority order. @@ -32,7 +50,9 @@ async def reserve_account( ): """Reserve an account and return ``(lease, selected_mode_id)``. - Returns ``(None, original_mode_id)`` when no account is available. + Returns ``(None, original_mode_id)`` when no account is available. Under the + random strategy no on-demand refresh fallback is attempted — upstream quota + data is never probed. """ original_mode_id = int(spec.mode_id) @@ -52,6 +72,9 @@ async def _try_reserve(): if lease is not None: return lease, selected_mode_id + if current_strategy() == "random": + return None, original_mode_id + refresh_svc = get_refresh_service() if refresh_svc is not None: await refresh_svc.refresh_on_demand() diff --git a/app/products/anthropic/messages.py b/app/products/anthropic/messages.py index 31ef2f90e..42ba98a17 100644 --- a/app/products/anthropic/messages.py +++ b/app/products/anthropic/messages.py @@ -33,7 +33,7 @@ _quota_sync, _fail_sync, _parse_retry_codes, _feedback_kind, _log_task_exception, _configured_retry_codes, _should_retry_upstream, ) -from app.products._account_selection import reserve_account +from app.products._account_selection import reserve_account, selection_max_retries from app.products.openai._tool_sieve import ToolSieve @@ -310,7 +310,7 @@ async def create( raise RateLimitError("Account directory not initialised") directory = _acct_dir - max_retries = cfg.get_int("retry.max_retries", 1) + max_retries = selection_max_retries() retry_codes = _configured_retry_codes(cfg) timeout_s = cfg.get_float("chat.timeout", 120.0) msg_id = _make_msg_id() diff --git a/app/products/openai/chat.py b/app/products/openai/chat.py index 4eeab694a..facdd4f55 100644 --- a/app/products/openai/chat.py +++ b/app/products/openai/chat.py @@ -54,7 +54,7 @@ build_usage, ) from ._tool_sieve import ToolSieve -from app.products._account_selection import reserve_account +from app.products._account_selection import reserve_account, selection_max_retries def _to_chat_annotations(anns: list[dict]) -> list[dict]: @@ -470,7 +470,7 @@ async def completions( raise RateLimitError("Account directory not initialised") directory = _acct_dir - max_retries = cfg.get_int("retry.max_retries", 1) + max_retries = selection_max_retries() retry_codes = _configured_retry_codes(cfg) response_id = make_response_id() timeout_s = cfg.get_float("chat.timeout", 120.0) diff --git a/app/products/openai/responses.py b/app/products/openai/responses.py index 95fdc3e7c..d816c7a92 100644 --- a/app/products/openai/responses.py +++ b/app/products/openai/responses.py @@ -18,7 +18,7 @@ from app.control.model.registry import resolve as resolve_model from app.control.account.enums import FeedbackKind from app.dataplane.reverse.protocol.xai_chat import classify_line, StreamAdapter -from app.products._account_selection import reserve_account +from app.products._account_selection import reserve_account, selection_max_retries from .chat import _stream_chat, _extract_message, _resolve_image, _quota_sync, _fail_sync, _parse_retry_codes, _feedback_kind, _log_task_exception, _upstream_body_excerpt from .chat import _configured_retry_codes, _should_retry_upstream @@ -247,7 +247,7 @@ async def create( raise RateLimitError("Account directory not initialised") directory = _acct_dir - max_retries = cfg.get_int("retry.max_retries", 1) + max_retries = selection_max_retries() retry_codes = _configured_retry_codes(cfg) response_id = make_resp_id("resp") reasoning_id = make_resp_id("rs") diff --git a/app/products/web/admin/__init__.py b/app/products/web/admin/__init__.py index 529452a4c..61494f8d9 100644 --- a/app/products/web/admin/__init__.py +++ b/app/products/web/admin/__init__.py @@ -206,6 +206,7 @@ async def get_storage_mode(): @router.get("/status", tags=[_TAG_ADMIN_SYSTEM]) async def runtime_status(): from app.dataplane.account import _directory + from app.dataplane.account.selector import current_strategy if _directory is None: raise AppError( @@ -220,6 +221,7 @@ async def runtime_status(): "status": "ok", "size": _directory.size, "revision": _directory.revision, + "selection_strategy": current_strategy(), } ), media_type="application/json", diff --git a/app/statics/admin/account.html b/app/statics/admin/account.html index f8983e3fa..9107c5de0 100644 --- a/app/statics/admin/account.html +++ b/app/statics/admin/account.html @@ -782,7 +782,7 @@