fix: add concurrency controls for multiple sessions using the same provider (#1069)

* fix: add concurrency controls for multiple sessions using the same provider Adds OS-inspired scheduling primitives (from HiveMind/AIMD research) to prevent concurrent sessions from the same provider from experiencing slowdowns, random stops, and cascading failures. Server-side: - Health check skips OpenCode restart when sessions are actively busy — a busy server under concurrent load can fail the health check timeout without being dead. Staleness guard forces restart if unhealthy+busy persists >2 minutes. - Upstream SSE stall timeout scaled from 20s to 60s to avoid unnecessary reconnections when multiple sessions are waiting for LLM responses. Client-side (HiveMind primitives, arXiv:2604.17111): - Transparent retry with exponential backoff (1s→2s→4s, max 32s) for 429/502/503/504 errors — the #1 most effective primitive from the paper. - Circuit breaker: opens after 3 consecutive retryable errors, cooldown doubles each trip (30s→60s→120s, capped 128s), matching TCP AIMD. - Per-provider session tracking with TTL eviction (1h idle sweep). - Fetch-level retry gated on AbortError/TypeError only (not DNS failures). Refs github-code-review skill findings (all 8 issues resolved). * fix: use definite assignment assertion for response variable Fixes TS2454: Variable 'response' is used before being assigned in strict mode. The for-loop body always assigns it on every path that reaches the post-loop code, but TS can't prove that. * fix: add cleanupSession to error paths and remove unreachable code P1 fixes (Greptile review): - cleanupSession called on fetch error throw path - cleanupSession called on non-retryable HTTP error throw path - Removed unreachable post-loop code (loop always terminates via return or throw) Adds explicit post-loop throw to satisfy TypeScript strict return check. * fix: address Greptile review feedback on concurrent session controls Removes client-side session tracking that leaked on normal completion paths. The session tracking was redundant — the server-side health check already reads from sessionRuntime.getSessionActivitySnapshot() for busy-session detection. Changes: - Remove activeSessions Set and all session-tracking functions from provider-tracker - Remove trackSessionStarted/cleanupSession calls from client.ts - Remove unreachable (response as Response) block after retry loop - Make upstreamStallTimeoutMs conditional: 60s when >1 sessions, 20s otherwise Refs #1069 * fix: enforce dynamic concurrency safeguards --------- Co-authored-by: Bohdan Triapitsyn <artmore@protonmail.com>
2026-05-01 14:57:31 +03:00
parent f9094bc3cc
commit a9ee499ae4
8 changed files with 316 additions and 39 deletions
@@ -15,12 +15,25 @@ import type {
 import type { PermissionRequest } from "@/types/permission";
 import type { QuestionRequest } from "@/types/question";
 import { waitForWorktreeBootstrap } from "@/lib/worktrees/worktreeBootstrap";
+import {
+  assertProviderCircuitClosed,
+  recordProviderSuccess,
+  recordProviderError,
+  shouldRetry,
+  getRetryDelayMs,
+} from "./provider-tracker";

 // Use relative path by default (works with both dev and nginx proxy server)
 // Can be overridden with VITE_OPENCODE_URL for absolute URLs in special deployments
 const DEFAULT_BASE_URL = import.meta.env.VITE_OPENCODE_URL || "/api";
 const ABSOLUTE_URL_PATTERN = /^[a-zA-Z][a-zA-Z\d+\-.]*:\/\//;

+const isRetryableFetchError = (error: unknown): boolean => {
+  if (error instanceof DOMException && error.name === 'AbortError') return true;
+  if (error instanceof TypeError) return true;
+  return false;
+};
+
 const ensureAbsoluteBaseUrl = (candidate: string): string => {
  const normalized = typeof candidate === "string" && candidate.trim().length > 0 ? candidate.trim() : "/api";

@@ -725,39 +738,58 @@ class OpencodeService {
      });
    }

-    let response: Response;
-    try {
-      response = await fetch(url.toString(), {
-        method: 'POST',
-        headers: {
-          'content-type': 'application/json',
-          accept: 'application/json',
-        },
-        body: JSON.stringify({
-          model: {
-            providerID: params.providerID,
-            modelID: params.modelID,
-          },
-          agent: params.agent,
-          variant: params.variant,
-          ...(params.messageId ? { messageID: params.messageId } : {}),
-          ...(params.format ? { format: params.format } : {}),
-          parts,
-        }),
-      });
-    } catch (error) {
-      console.error('[git-generation][browser] prompt_async request failed before response', {
-        sessionId: params.id,
-        url: url.toString(),
-        directory: this.currentDirectory,
-        hasFormat: Boolean(params.format),
-        message: error instanceof Error ? error.message : String(error),
-        error,
-      });
-      throw error;
-    }
+    assertProviderCircuitClosed(params.providerID);
+
+    let response!: Response;
+
+    for (let attempt = 0; attempt < 3; attempt++) {
+      try {
+        response = await fetch(url.toString(), {
+          method: 'POST',
+          headers: {
+            'content-type': 'application/json',
+            accept: 'application/json',
+          },
+          body: JSON.stringify({
+            model: {
+              providerID: params.providerID,
+              modelID: params.modelID,
+            },
+            agent: params.agent,
+            variant: params.variant,
+            ...(params.messageId ? { messageID: params.messageId } : {}),
+            ...(params.format ? { format: params.format } : {}),
+            parts,
+          }),
+        });
+      } catch (error) {
+        if (attempt < 2 && isRetryableFetchError(error)) {
+          const delay = getRetryDelayMs(attempt);
+          console.warn(
+            `[prompt] fetch failed for ${params.providerID}/${params.modelID} (attempt ${attempt + 1}/3), retrying in ${delay}ms`,
+            (error as Error)?.message
+          );
+          await new Promise((resolve) => setTimeout(resolve, delay));
+          continue;
+        }
+        recordProviderError(params.providerID);
+        throw error;
+      }
+
+      if (response.ok) {
+        recordProviderSuccess(params.providerID);
+        return tempMessageId;
+      }
+
+      if (shouldRetry(params.providerID, response.status, attempt)) {
+        const delay = getRetryDelayMs(attempt);
+        console.warn(
+          `[prompt] ${response.status} for ${params.providerID}/${params.modelID} (attempt ${attempt + 1}/3), retrying in ${delay}ms`
+        );
+        await new Promise((resolve) => setTimeout(resolve, delay));
+        continue;
+      }

-    if (!response.ok) {
      let detail = '';
      try {
        detail = await response.text();
@@ -765,12 +797,13 @@ class OpencodeService {
        // ignore
      }
      const suffix = detail && detail.trim().length > 0 ? `: ${detail.trim()}` : '';
-      throw new Error(`Failed to send message (${response.status})${suffix}`);
+      const error = new Error(`Failed to send message (${response.status})${suffix}`);
+      recordProviderError(params.providerID, response.status);
+      throw error;
    }
-
-    // Return temporary ID for optimistic UI
-    // Real messageID will come from server via SSE events
-    return tempMessageId;
+    // Defensive fallback — all loop paths return/throw, but TypeScript
+    // control flow analysis cannot prove exhaustiveness without this.
+    throw new Error('Failed to send message after retries');
  }

  async sendCommand(params: {
@@ -0,0 +1,133 @@
+/**
+ * Provider Circuit-Breaker & Retry Tracker
+ *
+ * Tracks per-provider error state to enable:
+ * - Transparent retry with exponential backoff for transient errors
+ * - Circuit breaking (pause requests to a provider during error storms)
+ *
+ * Inspired by HiveMind (arXiv:2604.17111) OS-inspired scheduling primitives.
+ */
+
+const DEFAULT_CIRCUIT_BREAK_THRESHOLD = 3
+const DEFAULT_CIRCUIT_COOLDOWN_MS = 30_000
+const DEFAULT_RETRY_BASE_DELAY_MS = 1000
+const DEFAULT_RETRY_MAX_DELAY_MS = 32_000
+const DEFAULT_RETRY_MAX_ATTEMPTS = 3
+const PROVIDER_EVICTION_TTL_MS = 60 * 60 * 1000
+const PROVIDER_EVICTION_INTERVAL_MS = 10 * 60 * 1000
+
+const RETRYABLE_STATUS_CODES = new Set([429, 502, 503, 504])
+
+type ProviderState = {
+  consecutiveErrors: number
+  lastErrorAt: number
+  circuitOpen: boolean
+  circuitOpenAt: number
+  circuitCooldownMs: number
+}
+
+const providers = new Map<string, ProviderState>()
+
+function evictStaleProviders(): void {
+  const now = Date.now()
+  for (const [providerID, state] of providers) {
+    if (
+      state.consecutiveErrors === 0 &&
+      now - state.lastErrorAt > PROVIDER_EVICTION_TTL_MS
+    ) {
+      providers.delete(providerID)
+    }
+  }
+}
+
+if (typeof setInterval !== 'undefined') {
+  setInterval(evictStaleProviders, PROVIDER_EVICTION_INTERVAL_MS)
+}
+
+function getOrCreateProvider(providerID: string): ProviderState {
+  let state = providers.get(providerID)
+  if (!state) {
+    state = {
+      consecutiveErrors: 0,
+      lastErrorAt: 0,
+      circuitOpen: false,
+      circuitOpenAt: 0,
+      circuitCooldownMs: DEFAULT_CIRCUIT_COOLDOWN_MS,
+    }
+    providers.set(providerID, state)
+  }
+  return state
+}
+
+export function recordProviderSuccess(providerID: string): void {
+  if (!providerID) return
+  const state = providers.get(providerID)
+  if (!state) return
+  state.consecutiveErrors = 0
+  state.lastErrorAt = 0
+}
+
+export function recordProviderError(providerID: string, status?: number): void {
+  if (!providerID) return
+  const state = getOrCreateProvider(providerID)
+  state.consecutiveErrors += 1
+  state.lastErrorAt = Date.now()
+
+  if (
+    isCircuitBreakerStatus(status) &&
+    state.consecutiveErrors >= DEFAULT_CIRCUIT_BREAK_THRESHOLD
+  ) {
+    state.circuitOpen = true
+    state.circuitOpenAt = Date.now()
+    console.warn(
+      `[provider-tracker] Circuit opened for ${providerID} after ${state.consecutiveErrors} consecutive errors`
+    )
+  }
+}
+
+function isCircuitBreakerStatus(status?: number): boolean {
+  return status !== undefined && RETRYABLE_STATUS_CODES.has(status)
+}
+
+export function isCircuitOpen(providerID: string): boolean {
+  const state = providers.get(providerID)
+  if (!state?.circuitOpen) return false
+
+  const elapsed = Date.now() - state.circuitOpenAt
+  if (elapsed >= state.circuitCooldownMs) {
+    state.circuitOpen = false
+    state.consecutiveErrors = 0
+    state.circuitCooldownMs = Math.min(
+      state.circuitCooldownMs * 2,
+      DEFAULT_RETRY_MAX_DELAY_MS * 4
+    )
+    return false
+  }
+
+  return true
+}
+
+export function shouldRetry(providerID: string, status: number, attempt: number): boolean {
+  if (!RETRYABLE_STATUS_CODES.has(status)) return false
+  if (attempt >= DEFAULT_RETRY_MAX_ATTEMPTS - 1) return false
+  if (isCircuitOpen(providerID)) return false
+  return true
+}
+
+export function assertProviderCircuitClosed(providerID: string): void {
+  if (!providerID || !isCircuitOpen(providerID)) return
+  throw new Error(`Provider ${providerID} is temporarily unavailable after repeated errors. Please retry shortly.`)
+}
+
+export function getRetryDelayMs(attempt: number): number {
+  const delay = DEFAULT_RETRY_BASE_DELAY_MS * 2 ** attempt
+  return Math.min(delay, DEFAULT_RETRY_MAX_DELAY_MS)
+}
+
+export function resetCircuit(providerID: string): void {
+  const state = providers.get(providerID)
+  if (!state) return
+  state.consecutiveErrors = 0
+  state.circuitOpen = false
+  state.circuitCooldownMs = DEFAULT_CIRCUIT_COOLDOWN_MS
+}
@@ -35,6 +35,8 @@ import {
  createGlobalUiEventBroadcaster,
  createGlobalMessageStreamHub,
  createMessageStreamWsRuntime,
+  DEFAULT_UPSTREAM_STALL_TIMEOUT_MS,
+  UPSTREAM_STALL_TIMEOUT_CONCURRENT_MS,
 } from './lib/event-stream/index.js';
 import { createFsSearchRuntime as createFsSearchRuntimeFactory } from './lib/fs/search.js';
 import { createOpenCodeLifecycleRuntime } from './lib/opencode/lifecycle.js';
@@ -381,6 +383,17 @@ const sessionRuntime = createSessionRuntime({
  broadcastEvent: broadcastGlobalUiEvent,
 });

+const getActiveSessionCount = () => {
+  const snapshot = sessionRuntime.getSessionActivitySnapshot();
+  return Object.values(snapshot).filter((entry) => entry.type === 'busy').length;
+};
+
+const getUpstreamStallTimeoutMs = () => (
+  getActiveSessionCount() > 1
+    ? UPSTREAM_STALL_TIMEOUT_CONCURRENT_MS
+    : DEFAULT_UPSTREAM_STALL_TIMEOUT_MS
+);
+
 const projectConfigRuntime = createProjectConfigRuntime({
  fsPromises,
  path,
@@ -655,6 +668,7 @@ const setAutoAcceptSession = (...args) => notificationTriggerRuntime.setAutoAcce
 const globalMessageStreamHub = createGlobalMessageStreamHub({
  buildOpenCodeUrl,
  getOpenCodeAuthHeaders,
+  upstreamStallTimeoutMs: getUpstreamStallTimeoutMs,
 });

 const openCodeWatcherRuntime = createOpenCodeWatcherRuntime({
@@ -881,6 +895,7 @@ const openCodeLifecycleRuntime = createOpenCodeLifecycleRuntime({
  buildAugmentedPath,
  buildManagedOpenCodePath,
  getManagedOpenCodeShellEnvSnapshot: getLoginShellEnvSnapshot,
+  getActiveSessionCount,
 });

 const restartOpenCode = (...args) => openCodeLifecycleRuntime.restartOpenCode(...args);
@@ -1188,6 +1203,7 @@ async function main(options = {}) {
    globalEventHub: globalMessageStreamHub,
    processForwardedEventPayload,
    messageStreamWsClients: uiNotificationWsClients,
+    upstreamStallTimeoutMs: getUpstreamStallTimeoutMs,
    terminalHeartbeatIntervalMs: TERMINAL_INPUT_WS_HEARTBEAT_INTERVAL_MS,
    terminalRebindWindowMs: TERMINAL_INPUT_WS_REBIND_WINDOW_MS,
    terminalMaxRebindsPerWindow: TERMINAL_INPUT_WS_MAX_REBINDS_PER_WINDOW,
@@ -20,5 +20,6 @@ export {
 export {
  DEFAULT_UPSTREAM_RECONNECT_DELAY_MS,
  DEFAULT_UPSTREAM_STALL_TIMEOUT_MS,
+  UPSTREAM_STALL_TIMEOUT_CONCURRENT_MS,
  createUpstreamSseReader,
 } from './upstream-reader.js';
@@ -1,8 +1,14 @@
 import { parseSseEventEnvelope } from './protocol.js';

 export const DEFAULT_UPSTREAM_STALL_TIMEOUT_MS = 20_000;
+export const UPSTREAM_STALL_TIMEOUT_CONCURRENT_MS = DEFAULT_UPSTREAM_STALL_TIMEOUT_MS * 3;
 export const DEFAULT_UPSTREAM_RECONNECT_DELAY_MS = 250;

+function resolveTimeoutMs(value, fallback) {
+  const resolved = typeof value === 'function' ? value() : value;
+  return Number.isFinite(resolved) ? resolved : fallback;
+}
+
 function waitForReconnectDelay(ms, signal) {
  if (signal?.aborted) {
    return Promise.resolve();
@@ -76,14 +82,15 @@ export function createUpstreamSseReader({
        };
        const resetStallTimer = () => {
          clearStallTimer();
-          if (stallTimeoutMs <= 0) {
+          const currentStallTimeoutMs = resolveTimeoutMs(stallTimeoutMs, DEFAULT_UPSTREAM_STALL_TIMEOUT_MS);
+          if (currentStallTimeoutMs <= 0) {
            return;
          }

          stallTimer = setTimeout(() => {
            abortReason = 'upstream_stalled';
            controller.abort();
-          }, stallTimeoutMs);
+          }, currentStallTimeoutMs);
        };

        try {
@@ -117,6 +117,51 @@ describe('createUpstreamSseReader', () => {
    expect(reader.getLastEventId()).toBe('evt-2');
  });

+  it('resolves the stall timeout for each upstream read window', async () => {
+    const events = [];
+    let attempt = 0;
+    let currentTimeout = 10;
+    let reader;
+
+    reader = createUpstreamSseReader({
+      buildUrl: () => 'http://127.0.0.1:4096/global/event',
+      stallTimeoutMs: () => currentTimeout,
+      reconnectDelayMs: 0,
+      fetchImpl: async (_url, options) => {
+        attempt += 1;
+
+        if (attempt === 1) {
+          currentTimeout = 60;
+          return createSseResponse({
+            signal: options.signal,
+            holdOpen: true,
+            blocks: [
+              'id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n',
+            ],
+          });
+        }
+
+        return createSseResponse({
+          signal: options.signal,
+          blocks: [
+            'id: evt-2\ndata: {"type":"session.updated","properties":{}}\n\n',
+          ],
+        });
+      },
+      onEvent(event) {
+        events.push(event.eventId);
+        if (event.eventId === 'evt-2') {
+          reader.stop();
+        }
+      },
+    });
+
+    await reader.start();
+
+    expect(events).toEqual(['evt-1', 'evt-2']);
+    expect(attempt).toBe(2);
+  });
+
  it('reports unavailable upstream responses and continues reconnecting until stopped', async () => {
    const errors = [];
    let attempt = 0;
@@ -25,6 +25,7 @@ export const createOpenCodeLifecycleRuntime = (deps) => {
    buildAugmentedPath,
    buildManagedOpenCodePath,
    getManagedOpenCodeShellEnvSnapshot,
+    getActiveSessionCount = () => 0,
  } = deps;

  const killProcessOnPort = (port) => {
@@ -796,15 +797,51 @@ export const createOpenCodeLifecycleRuntime = (deps) => {
   * Perform an immediate (one-shot) health check and restart OpenCode if it's
   * not healthy.  Callers on the SSE / WS proxy path use this to trigger
   * recovery without waiting for the next periodic interval (up to 15 s).
+   *
+   * Skips restart when sessions are actively busy — a busy server under
+   * concurrent load can fail the health check timeout without actually
+   * being dead (the health endpoint competes with LLM work).
+   * Forces restart if sessions stay "busy" and the server stays unhealthy
+   * for over 2 minutes (staleness guard against stuck session state).
   */
+  const STALE_BUSY_GRACE_MS = 2 * 60 * 1000;
+  let lastUnhealthyWithBusySessionsAt = 0;
+
+  const shouldSkipRestartForBusySessions = () => {
+    const activeCount = getActiveSessionCount();
+    if (activeCount === 0) {
+      lastUnhealthyWithBusySessionsAt = 0;
+      return false;
+    }
+
+    const now = Date.now();
+    if (!lastUnhealthyWithBusySessionsAt) {
+      lastUnhealthyWithBusySessionsAt = now;
+      return true;
+    }
+
+    if (now - lastUnhealthyWithBusySessionsAt >= STALE_BUSY_GRACE_MS) {
+      console.warn(
+        `[lifecycle] OpenCode unhealthy with ${activeCount} busy session(s) for > 2 min — forcing restart`
+      );
+      lastUnhealthyWithBusySessionsAt = 0;
+      return false;
+    }
+
+    return true;
+  };
+
  const triggerHealthCheck = async () => {
    if (!state.openCodeProcess || state.isShuttingDown || state.isRestartingOpenCode) return;

    try {
      const healthy = await isOpenCodeProcessHealthy();
      if (!healthy) {
+        if (shouldSkipRestartForBusySessions()) return;
        console.log('[lifecycle] immediate health check: OpenCode not healthy, restarting...');
        await restartOpenCode();
+      } else {
+        lastUnhealthyWithBusySessionsAt = 0;
      }
    } catch (error) {
      console.error(`[lifecycle] immediate health check error: ${error.message}`);
@@ -822,8 +859,11 @@ export const createOpenCodeLifecycleRuntime = (deps) => {
      try {
        const healthy = await isOpenCodeProcessHealthy();
        if (!healthy) {
+          if (shouldSkipRestartForBusySessions()) return;
          console.log('OpenCode process not running, restarting...');
          await restartOpenCode();
+        } else {
+          lastUnhealthyWithBusySessionsAt = 0;
        }
      } catch (error) {
        console.error(`Health check error: ${error.message}`);
@@ -24,6 +24,7 @@ export const createStartupPipelineRuntime = (dependencies) => {
      processForwardedEventPayload,
      messageStreamWsClients,
      triggerHealthCheck,
+      upstreamStallTimeoutMs,
      terminalHeartbeatIntervalMs,
      terminalRebindWindowMs,
      terminalMaxRebindsPerWindow,
@@ -80,6 +81,7 @@ export const createStartupPipelineRuntime = (dependencies) => {
      processForwardedEventPayload,
      wsClients: messageStreamWsClients,
      triggerHealthCheck,
+      upstreamStallTimeoutMs,
    });

    setupProxy(app);