diff --git a/desktop/src/electron/config/ui-config.ts b/desktop/src/electron/config/ui-config.ts index f24fad77..2ed907a6 100644 --- a/desktop/src/electron/config/ui-config.ts +++ b/desktop/src/electron/config/ui-config.ts @@ -6,9 +6,11 @@ import path from 'path' import { getPaths } from '@/electron/globals' import { MODULAR_DEFAULT_LOG_LEVEL, + MODULAR_DEFAULT_PROXY_RESPONSE_TIMEOUT_MINUTES, isModularLogLevel, type ModularLogLevel } from '@/shared/constants/modular-runtime' +import { isValidProxyResponseTimeoutMinutes } from '@/shared/utils/proxy-response-timeout' interface UiConfig { /** When true, first-run onboarding has not been completed or explicitly dismissed. */ @@ -17,12 +19,15 @@ interface UiConfig { modularLogLevel: ModularLogLevel /** macOS only: the one-time privileged-helper setup (register the SMAppService daemon + configure the Application Firewall) has completed. Gates the first-run admin prompt; left false until the daemon is enabled and firewall configuration succeeds, so an approval-pending launch retries next time. */ macHelperSetupComplete: boolean + /** Minutes passed to the broker as `--proxy-response-timeout` at spawn. 0 means wait indefinitely (no automatic failover). Takes effect on the next service restart. */ + proxyResponseTimeoutMinutes: number } const DEFAULTS: UiConfig = { firstRun: true, modularLogLevel: MODULAR_DEFAULT_LOG_LEVEL, - macHelperSetupComplete: false + macHelperSetupComplete: false, + proxyResponseTimeoutMinutes: MODULAR_DEFAULT_PROXY_RESPONSE_TIMEOUT_MINUTES } let config: UiConfig = { ...DEFAULTS } @@ -121,3 +126,15 @@ export function setMacHelperSetupComplete(value: boolean): void { config.macHelperSetupComplete = value save() } + +export function getProxyResponseTimeoutMinutes(): number { + // Guard against a hand-edited / legacy config value that isn't valid. + return isValidProxyResponseTimeoutMinutes(config.proxyResponseTimeoutMinutes) + ? config.proxyResponseTimeoutMinutes + : MODULAR_DEFAULT_PROXY_RESPONSE_TIMEOUT_MINUTES +} + +export function setProxyResponseTimeoutMinutes(value: number): void { + config.proxyResponseTimeoutMinutes = value + save() +} diff --git a/desktop/src/electron/connector/index.ts b/desktop/src/electron/connector/index.ts index c35d991f..9cd5da34 100644 --- a/desktop/src/electron/connector/index.ts +++ b/desktop/src/electron/connector/index.ts @@ -3,7 +3,7 @@ import { BrowserWindow } from 'electron' import { createStructuredLogger } from '@/shared/utils/log' -import { getModularLogLevel } from '@/electron/config/ui-config' +import { getModularLogLevel, getProxyResponseTimeoutMinutes } from '@/electron/config/ui-config' import getErrorString from '@/shared/utils/get-error-string' import { getModularSupervisor, @@ -141,6 +141,9 @@ export const initializeConnector = async (): Promise => { // Seed the persisted log level so spawn passes the right `--log-level`. const supervisor = getModularSupervisor() supervisor.setLogLevel(getModularLogLevel()) + // Same for the broker's --proxy-response-timeout (spawn-arg only, no + // live JSON-RPC equivalent — see setProxyResponseTimeout's doc comment). + supervisor.setProxyResponseTimeout(getProxyResponseTimeoutMinutes()) ensureBrokerCrashHandler() supervisor.start() weSpawned = true diff --git a/desktop/src/electron/ipc/service.ipc.ts b/desktop/src/electron/ipc/service.ipc.ts index b7b1ebcd..663d0957 100644 --- a/desktop/src/electron/ipc/service.ipc.ts +++ b/desktop/src/electron/ipc/service.ipc.ts @@ -16,12 +16,18 @@ import { destroyConnector, restartConnector } from '@/electron/connector' -import { getModularLogLevel, setModularLogLevel } from '@/electron/config/ui-config' +import { + getModularLogLevel, + setModularLogLevel, + getProxyResponseTimeoutMinutes, + setProxyResponseTimeoutMinutes +} from '@/electron/config/ui-config' import { getModularSupervisor, readCliBinManifest } from '@/electron/service-bridge/modular-supervisor' import { modularShippedBinaryBaseNames } from '@/shared/constants/modular-binaries' +import { isValidProxyResponseTimeoutMinutes } from '@/shared/utils/proxy-response-timeout' const LICENSE_FILE = 'LICENSE' const THIRD_PARTY_LICENSE_FILE = 'THIRD_PARTY_NOTICES.md' @@ -98,6 +104,22 @@ export function registerServiceIpc(): void { getModularSupervisor().setLogLevel(payload.level) }) + safeHandle('service:get-proxy-response-timeout', async () => { + return getProxyResponseTimeoutMinutes() + }) + + safeHandle('service:set-proxy-response-timeout', async (_event, payload) => { + if (!isValidProxyResponseTimeoutMinutes(payload.minutes)) { + throw new Error('Proxy response timeout must be a non-negative number of minutes') + } + // Spawn-arg only (no JSON-RPC equivalent to fan out live like log level) — + // persist for the next broker restart. Reflected in the supervisor's + // in-memory value too, so a restart triggered right after saving already + // picks up the new value even before the renderer refetches it. + setProxyResponseTimeoutMinutes(payload.minutes) + getModularSupervisor().setProxyResponseTimeout(payload.minutes) + }) + safeHandle('service:open-log-file', async () => { const logPath = getStructuredLogFilePath() if (!logPath || !fs.existsSync(logPath)) return diff --git a/desktop/src/electron/service-bridge/modular-supervisor.ts b/desktop/src/electron/service-bridge/modular-supervisor.ts index 943d0f84..4dcd24f6 100644 --- a/desktop/src/electron/service-bridge/modular-supervisor.ts +++ b/desktop/src/electron/service-bridge/modular-supervisor.ts @@ -33,11 +33,16 @@ import { parseClusterNodes, parseInvite, parseNodeIdentity } from './cluster-jso import { startNodeInfoPoller, stopNodeInfoPoller } from './node-info-poller' import { MODULAR_DEFAULT_LOG_LEVEL, + MODULAR_DEFAULT_PROXY_RESPONSE_TIMEOUT_MINUTES, MODULAR_INVITE_STATUS_POLL_INTERVAL_MS, MODULAR_MODEL_ACTION_TIMEOUT_MS, isModularLogLevel, type ModularLogLevel } from '@/shared/constants/modular-runtime' +import { + isValidProxyResponseTimeoutMinutes, + proxyResponseTimeoutArg +} from '@/shared/utils/proxy-response-timeout' import { listManualNodeEntries } from './manual-nodes-store' import { MODULAR_RUNTIME_BINARIES, @@ -350,6 +355,11 @@ class ModularSupervisor { // connector seeds it via setLogLevel() before start() so spawn args use it. // This default only applies if start() runs before the connector seeds. private logLevel: ModularLogLevel = MODULAR_DEFAULT_LOG_LEVEL + // Same pattern as logLevel above, for the broker's `--proxy-response-timeout` + // (see ui-config's `proxyResponseTimeoutMinutes`). Unlike log level this has + // no live JSON-RPC fan-out — the broker only reads it at spawn — so a change + // takes effect on the next service restart, not immediately. + private proxyResponseTimeoutMinutes: number = MODULAR_DEFAULT_PROXY_RESPONSE_TIMEOUT_MINUTES private brokerReady = false private isReady = false // True only while stop() is intentionally tearing down the subprocess tree. @@ -739,6 +749,21 @@ class ModularSupervisor { return this.logLevel } + /** + * Update the in-memory value used for the broker's next + * `--proxy-response-timeout` spawn arg. No live effect on a running broker + * (it is a spawn-time flag, not a JSON-RPC-settable one) — the caller is + * expected to restart the service for this to take effect. + */ + setProxyResponseTimeout(minutes: number): void { + if (!isValidProxyResponseTimeoutMinutes(minutes)) return + this.proxyResponseTimeoutMinutes = minutes + } + + getProxyResponseTimeout(): number { + return this.proxyResponseTimeoutMinutes + } + private validateRequiredBinaries(): void { for (const definition of MODULAR_RUNTIME_BINARIES) { if (definition.optional) continue @@ -838,6 +863,10 @@ class ModularSupervisor { // streams, and fans its schedule:priority out to the proxies via // node/set-priority (all broker-internal). passPath('--scheduler-path', 'job-scheduler') + args.push( + '--proxy-response-timeout', + proxyResponseTimeoutArg(this.proxyResponseTimeoutMinutes) + ) return [...args, ...this.logLevelArgs()] } diff --git a/desktop/src/preload/api/service.api.ts b/desktop/src/preload/api/service.api.ts index 8bb05bef..5daa6c16 100644 --- a/desktop/src/preload/api/service.api.ts +++ b/desktop/src/preload/api/service.api.ts @@ -14,6 +14,9 @@ export interface IServiceApi { restart(): Promise getLogLevel(): Promise setLogLevel(level: ModularLogLevel): Promise + /** Minutes; applies on the next service restart. */ + getProxyResponseTimeout(): Promise + setProxyResponseTimeout(minutes: number): Promise openLogFile(): Promise openLogDir(): Promise openLicense(): Promise @@ -31,6 +34,12 @@ export const serviceApi: IServiceApi = { invokeAndUnwrap(ipcRenderer.invoke('service:get-log-level')), setLogLevel: level => invokeAndUnwrap(ipcRenderer.invoke('service:set-log-level', { level })), + getProxyResponseTimeout: () => + invokeAndUnwrap(ipcRenderer.invoke('service:get-proxy-response-timeout')), + setProxyResponseTimeout: minutes => + invokeAndUnwrap( + ipcRenderer.invoke('service:set-proxy-response-timeout', { minutes }) + ), openLogFile: () => invokeAndUnwrap(ipcRenderer.invoke('service:open-log-file')), openLogDir: () => invokeAndUnwrap(ipcRenderer.invoke('service:open-log-dir')), openLicense: () => invokeAndUnwrap(ipcRenderer.invoke('service:open-license')), diff --git a/desktop/src/shared/constants/modular-runtime.ts b/desktop/src/shared/constants/modular-runtime.ts index 9b833e9a..cfd042da 100644 --- a/desktop/src/shared/constants/modular-runtime.ts +++ b/desktop/src/shared/constants/modular-runtime.ts @@ -26,6 +26,19 @@ export function isModularLogLevel(value: string): value is ModularLogLevel { export const MODULAR_DEFAULT_LOG_LEVEL: ModularLogLevel = 'warn' +// Default for the broker's `--proxy-response-timeout`, in minutes. Mirrors the +// broker's own hardcoded default (5*time.Minute in +// services/nvpair-ui-broker/main.go) so a user who never touches this setting +// gets byte-for-byte the same spawn args as before this setting existed. +// +// This is how long the broker tells ollama-proxy to wait for a forwarded +// request's response headers before giving up and failing over to another +// node — see --response-timeout in services/ollama-proxy/main.go. 0 disables +// the timeout entirely (wait indefinitely): only safe against a backend +// trusted to eventually respond or fail on its own, since a genuinely wedged +// backend then hangs the request forever with no automatic failover. +export const MODULAR_DEFAULT_PROXY_RESPONSE_TIMEOUT_MINUTES = 5 + // Local `/v1/node-info` HTTP poll. This is the single sanctioned HTTP exception // for the modular bridge: the broker does not yet expose rich per-node // telemetry, so Electron polls each discovered node's `/v1/node-info` endpoint diff --git a/desktop/src/shared/types/ipc-channels.ts b/desktop/src/shared/types/ipc-channels.ts index 289319d3..55fb8fd8 100644 --- a/desktop/src/shared/types/ipc-channels.ts +++ b/desktop/src/shared/types/ipc-channels.ts @@ -89,6 +89,9 @@ export interface IpcChannelMap { 'service:restart': { request: void; response: void } 'service:get-log-level': { request: void; response: ModularLogLevel } 'service:set-log-level': { request: { level: ModularLogLevel }; response: void } + /** Minutes; applies on the next service restart (spawn-arg only, no live fan-out). */ + 'service:get-proxy-response-timeout': { request: void; response: number } + 'service:set-proxy-response-timeout': { request: { minutes: number }; response: void } 'service:open-log-file': { request: void; response: void } 'service:open-log-dir': { request: void; response: void } 'service:get-versions': { request: void; response: ServiceVersions } diff --git a/desktop/src/shared/utils/proxy-response-timeout.ts b/desktop/src/shared/utils/proxy-response-timeout.ts new file mode 100644 index 00000000..9a1ac012 --- /dev/null +++ b/desktop/src/shared/utils/proxy-response-timeout.ts @@ -0,0 +1,28 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +/** + * Validation and CLI-arg formatting for the "Proxy response timeout" setting + * (persisted as minutes, passed to the broker as `--proxy-response-timeout`). + * + * The value is entered and stored as minutes rather than a raw Go duration + * string (e.g. "5m30s") so the UI never has to parse or round-trip Go's + * duration grammar — a plain non-negative number is unambiguous, and 0 maps + * directly onto the broker/proxy's own "0 disables the timeout" convention + * (see `services/nvpair-ui-broker/main.go`, `services/ollama-proxy/main.go`). + */ + +/** Non-negative, finite minutes — `0` means "wait indefinitely". */ +export function isValidProxyResponseTimeoutMinutes(value: number): boolean { + return Number.isFinite(value) && value >= 0 +} + +/** + * Render minutes as the Go duration string the broker's `--proxy-response- + * timeout` flag (a `time.Duration`) expects, e.g. `5` -> `"5m"`, `0` -> `"0m"`. + * Go's `time.ParseDuration` accepts fractional values with a unit (`"2.5m"`), + * so this is safe for non-integer minutes too. + */ +export function proxyResponseTimeoutArg(minutes: number): string { + return `${minutes}m` +} diff --git a/desktop/src/ui/components/ServiceSettings/ServiceSettings.tsx b/desktop/src/ui/components/ServiceSettings/ServiceSettings.tsx index 9eb9e88e..ee81a020 100644 --- a/desktop/src/ui/components/ServiceSettings/ServiceSettings.tsx +++ b/desktop/src/ui/components/ServiceSettings/ServiceSettings.tsx @@ -9,6 +9,7 @@ import { Flex, Stack, Text, + TextInput, type DropdownEntry } from '@nvidia/foundations-react-core' import { useConnectionStore } from '@/ui/stores/connection.store' @@ -16,9 +17,11 @@ import { Download, OpenInNew } from '@/ui/components/icons' import type { ServiceStatus } from '@/shared/types/ipc-channels' import { MODULAR_DEFAULT_LOG_LEVEL, + MODULAR_DEFAULT_PROXY_RESPONSE_TIMEOUT_MINUTES, MODULAR_LOG_LEVELS, type ModularLogLevel } from '@/shared/constants/modular-runtime' +import { isValidProxyResponseTimeoutMinutes } from '@/shared/utils/proxy-response-timeout' import getErrorString from '@/shared/utils/get-error-string' import { DismissibleTooltip } from '@/ui/components/DismissibleTooltip/DismissibleTooltip' import { isElectron } from '@/ui/api/bootstrap' @@ -101,6 +104,16 @@ export default function ServiceSettings() { const [error, setError] = useState(null) const [loading, setLoading] = useState(null) const [logLevel, setLogLevel] = useState(MODULAR_DEFAULT_LOG_LEVEL) + // Persisted value (minutes), and the input's own draft text — kept apart so + // typing an in-progress/invalid value never clobbers the saved one, and so + // "changed" can be judged against what is actually stored. + const [proxyTimeoutMinutes, setProxyTimeoutMinutes] = useState( + MODULAR_DEFAULT_PROXY_RESPONSE_TIMEOUT_MINUTES + ) + const [proxyTimeoutInput, setProxyTimeoutInput] = useState( + String(MODULAR_DEFAULT_PROXY_RESPONSE_TIMEOUT_MINUTES) + ) + const [savingProxyTimeout, setSavingProxyTimeout] = useState(false) const pollRef = useRef | null>(null) const [startingDemo, setStartingDemo] = useState(false) const setActiveTab = useOverviewUiStore(state => state.setActiveTab) @@ -140,6 +153,17 @@ export default function ServiceSettings() { .catch(() => {}) }, []) + useEffect(() => { + if (!isElectron) return + window.windowApi.service + .getProxyResponseTimeout() + .then(minutes => { + setProxyTimeoutMinutes(minutes) + setProxyTimeoutInput(String(minutes)) + }) + .catch(() => {}) + }, []) + const handleLogLevelChange = useCallback( async (level: ModularLogLevel) => { const prev = logLevel @@ -154,6 +178,28 @@ export default function ServiceSettings() { [logLevel] ) + const parsedProxyTimeout = Number(proxyTimeoutInput.trim()) + const proxyTimeoutValid = isValidProxyResponseTimeoutMinutes(parsedProxyTimeout) + const proxyTimeoutChanged = proxyTimeoutValid && parsedProxyTimeout !== proxyTimeoutMinutes + + const handleSaveProxyTimeout = useCallback(async () => { + if (!proxyTimeoutValid) { + setError('Enter a non-negative number of minutes (0 waits indefinitely).') + return + } + setSavingProxyTimeout(true) + setError(null) + try { + await window.windowApi.service.setProxyResponseTimeout(parsedProxyTimeout) + setProxyTimeoutMinutes(parsedProxyTimeout) + setProxyTimeoutInput(String(parsedProxyTimeout)) + } catch (err) { + setError(getErrorString(err)) + } finally { + setSavingProxyTimeout(false) + } + }, [parsedProxyTimeout, proxyTimeoutValid]) + const logLevelItems: DropdownEntry[] = useMemo( () => MODULAR_LOG_LEVELS.map(level => ({ @@ -374,6 +420,64 @@ export default function ServiceSettings() { )} + + + Proxy response timeout + {isElectron ? ( + + + + min + + + + ) : ( + + + + + + )} + + + How long the proxy waits for a node's response before failing + over to another node. Default is{' '} + {MODULAR_DEFAULT_PROXY_RESPONSE_TIMEOUT_MINUTES} minutes, matching + the broker's own default — leaving this unset changes nothing. + Set to 0 to wait indefinitely: useful for a very slow model, but it + means no automatic failover if a node hangs. Takes effect the next + time the service restarts. + + + diff --git a/desktop/tests/modular/service-startup-failure.test.ts b/desktop/tests/modular/service-startup-failure.test.ts index d769fc96..a72b70e3 100644 --- a/desktop/tests/modular/service-startup-failure.test.ts +++ b/desktop/tests/modular/service-startup-failure.test.ts @@ -10,6 +10,7 @@ const mocks = vi.hoisted(() => { const supervisor = { setLogLevel: vi.fn((_level: string): void => {}), + setProxyResponseTimeout: vi.fn((_minutes: number): void => {}), setOnBrokerCrash: vi.fn((_callback: (info: { code: number | null }) => void): void => {}), setOnReady: vi.fn((callback: () => void): void => { readyCallback = callback @@ -43,7 +44,8 @@ vi.mock('@/shared/utils/log', () => ({ })) vi.mock('@/electron/config/ui-config', () => ({ - getModularLogLevel: () => 'debug' + getModularLogLevel: () => 'debug', + getProxyResponseTimeoutMinutes: () => 5 })) vi.mock('@/electron/service-bridge/modular-supervisor', () => ({ diff --git a/services/nvpair-ui-broker/broker.go b/services/nvpair-ui-broker/broker.go index 0d189578..e4dd60a3 100644 --- a/services/nvpair-ui-broker/broker.go +++ b/services/nvpair-ui-broker/broker.go @@ -145,22 +145,23 @@ type ProxyStatusResult struct { // client connection in listen mode so future per-session caches (auth // tokens, watched-resource cursors, etc.) don't bleed across clients. type Broker struct { - codec *Codec - cancel context.CancelFunc - startedAt time.Time - nodeID string - scannerPath string - nodeInfoPath string - proxyPath string - lmstudioProxyPath string - workloadMgrPath string - errorsPath string - engineMgrPath string - manualNodesPath string - settingsPath string - clusterMgrPath string - schedulerPath string - clusterDir string + codec *Codec + cancel context.CancelFunc + startedAt time.Time + nodeID string + scannerPath string + nodeInfoPath string + proxyPath string + proxyResponseTimeout time.Duration + lmstudioProxyPath string + workloadMgrPath string + errorsPath string + engineMgrPath string + manualNodesPath string + settingsPath string + clusterMgrPath string + schedulerPath string + clusterDir string // Managed-port state is prepared before proxy startup and read by the proxy // supervisor/reader goroutines. Ollama commits its pending backend move after // its proxy reserves :11434; LM Studio moves through engine-manager first, @@ -341,6 +342,13 @@ type workerPaths struct { // trusted/). Threaded to every worker that does cluster-scoped inter-node // mTLS so they serve/dial pinned peers once this node joins a cluster. clusterDir string + // proxyResponseTimeout is passed to ollama-proxy as --response-timeout so + // it waits longer than its own 120s default for a real forwarded request's + // response headers (see spawnProxy). Zero means "wait indefinitely" + // (ollama-proxy's own zero-means-no-timeout semantics), explicitly passed + // through rather than omitted. Negative is invalid and skips the flag + // entirely, leaving ollama-proxy's own default in effect. + proxyResponseTimeout time.Duration } // NewBroker constructs a per-session broker. paths.scanner is required — @@ -362,30 +370,31 @@ func NewBroker(codec *Codec, paths workerPaths) *Broker { // its localNodeID stays in lockstep with what the broker stamps. nodeID := resolveLocalNodeID(paths.clusterDir) return &Broker{ - codec: codec, - startedAt: time.Now(), - nodeID: nodeID, - scannerPath: paths.scanner, - nodeInfoPath: paths.nodeInfo, - proxyPath: paths.proxy, - lmstudioProxyPath: paths.lmstudioProxy, - workloadMgrPath: paths.workloadMgr, - errorsPath: paths.errors, - engineMgrPath: paths.engineMgr, - manualNodesPath: paths.manualNodes, - settingsPath: paths.settings, - clusterMgrPath: paths.clusterMgr, - schedulerPath: paths.scheduler, - clusterDir: paths.clusterDir, - store: newDiscoveryStore(), - telemetry: newTelemetryCache(), - relayDir: relay.NewDirectory(), - regCache: relay.NewRegistrationCache(), - manualNodeKeys: make(map[string]string), - manualNodeStatuses: make(map[string]manualNodeStatusEntry), - workloads: workloadstore.New(), - ollamaPortReady: make(chan struct{}), - lmstudioPortReady: make(chan struct{}), + codec: codec, + startedAt: time.Now(), + nodeID: nodeID, + scannerPath: paths.scanner, + nodeInfoPath: paths.nodeInfo, + proxyPath: paths.proxy, + proxyResponseTimeout: paths.proxyResponseTimeout, + lmstudioProxyPath: paths.lmstudioProxy, + workloadMgrPath: paths.workloadMgr, + errorsPath: paths.errors, + engineMgrPath: paths.engineMgr, + manualNodesPath: paths.manualNodes, + settingsPath: paths.settings, + clusterMgrPath: paths.clusterMgr, + schedulerPath: paths.scheduler, + clusterDir: paths.clusterDir, + store: newDiscoveryStore(), + telemetry: newTelemetryCache(), + relayDir: relay.NewDirectory(), + regCache: relay.NewRegistrationCache(), + manualNodeKeys: make(map[string]string), + manualNodeStatuses: make(map[string]manualNodeStatusEntry), + workloads: workloadstore.New(), + ollamaPortReady: make(chan struct{}), + lmstudioPortReady: make(chan struct{}), } } @@ -650,6 +659,9 @@ func (b *Broker) spawnProxy() (supervisedHandle, error) { // ingress (and dial peers over mTLS) once this node is clustered; empty/ // absent certs leave it loopback-plaintext only. args = append(args, b.clusterDirArgs()...) + if b.proxyResponseTimeout >= 0 { + args = append(args, "--response-timeout", b.proxyResponseTimeout.String()) + } pp, err := startProxy("proxy", b.proxyPath, applog.LevelString(), b.relayDir, func(method string, params json.RawMessage) { b.forwardProxyNotificationForGeneration(generation, method, params) diff --git a/services/nvpair-ui-broker/main.go b/services/nvpair-ui-broker/main.go index ced0d783..028ea93c 100644 --- a/services/nvpair-ui-broker/main.go +++ b/services/nvpair-ui-broker/main.go @@ -15,6 +15,7 @@ import ( "path/filepath" "runtime" "syscall" + "time" "nvpair-shared/appdir" "nvpair-shared/applog" @@ -34,6 +35,7 @@ func main() { clusterMgrPath := flag.String("cluster-manager-path", "", "path to nvpair-cluster-manager binary (default: ./nvpair-cluster-manager in the current working directory)") schedulerPath := flag.String("scheduler-path", "", "path to nvpair-job-scheduler binary (default: ./nvpair-job-scheduler in the current working directory)") clusterDirFlag := flag.String("cluster-dir", "", "cluster config dir (node.crt/node.key + trusted/) the broker passes to its mDNS workers (nvpair-errors, nvpair-workload-manager, nvpair-node-info, nvpair-node-scanner, nvpair-manual-nodes) to enable cluster-scoped inter-node mTLS; defaults to the per-user Nvidia Corporation/Personal AI Router cluster/ dir, where nvpair-cluster-manager mints them") + proxyResponseTimeout := flag.Duration("proxy-response-timeout", 5*time.Minute, "response-header timeout the broker tells ollama-proxy to use for real forwarded requests (e.g. /v1/chat/completions). ollama-proxy's own standalone default is 120s; the broker asks for a longer 5m by default since a cold model load, a long prefill on a large context, or a tool-calling generation an engine only flushes once complete can easily exceed 120s on legitimate, healthy requests. 0 disables the timeout entirely (wait indefinitely) — only safe against a backend you trust to eventually respond or fail on its own, since a genuinely wedged backend will then hang the request forever with no automatic failover") showVersion := flag.Bool("version", false, "print version and exit") resolveLevel := applog.RegisterFlag(nil, slog.LevelInfo) flag.Parse() @@ -249,18 +251,19 @@ func main() { codec := NewCodec(transport) paths := workerPaths{ - scanner: resolvedScanner, - nodeInfo: resolvedNodeInfo, - proxy: resolvedProxy, - lmstudioProxy: resolvedLMStudioProxy, - workloadMgr: resolvedWorkloadMgr, - errors: resolvedErrors, - engineMgr: resolvedEngineMgr, - manualNodes: resolvedManualNodes, - settings: resolvedSettings, - clusterMgr: resolvedClusterMgr, - scheduler: resolvedScheduler, - clusterDir: clusterDir, + scanner: resolvedScanner, + nodeInfo: resolvedNodeInfo, + proxy: resolvedProxy, + lmstudioProxy: resolvedLMStudioProxy, + workloadMgr: resolvedWorkloadMgr, + errors: resolvedErrors, + engineMgr: resolvedEngineMgr, + manualNodes: resolvedManualNodes, + settings: resolvedSettings, + clusterMgr: resolvedClusterMgr, + scheduler: resolvedScheduler, + clusterDir: clusterDir, + proxyResponseTimeout: *proxyResponseTimeout, } if err := NewBroker(codec, paths).Serve(ctx); err != nil && ctx.Err() == nil { fatalf("broker error: %v", err) diff --git a/services/ollama-proxy/README.md b/services/ollama-proxy/README.md index 35f959b5..ed24cc47 100644 --- a/services/ollama-proxy/README.md +++ b/services/ollama-proxy/README.md @@ -28,6 +28,7 @@ ollama-proxy [flags] | `--ignore-persisted-port` | `false` | Use `--port` even when `proxy-port.json` contains a saved port (used by broker-managed startup) | | `--ipc` | *(empty — use stdio)* | Path to a Unix domain socket or Windows named pipe for IPC | | `--cluster-dir` | *(empty)* | Cluster trust directory (`node.crt`/`node.key` plus trusted pins). Enables the LAN mTLS inference ingress while this node is a cluster member; empty means no ingress and no peer candidates. | +| `--response-timeout` | `120s` | `ResponseHeaderTimeout` for a forwarded request (e.g. `/v1/chat/completions`) — how long the proxy waits for a backend Ollama/LM Studio node to start sending its response before failing over to the next candidate. Raise this (e.g. `--response-timeout 5m`) if a large or cold-loading local model needs longer than the default to produce its first response bytes. Accepts a Go duration string (`90s`, `5m`, `2m30s`). Does not affect the fast model-list/probe client, which stays fixed at the dial timeout. | | `--log-level` | *(`$NVPAIR_LOG_LEVEL`, else `info`)* | Initial log level: `debug`, `info`, `warn`, or `error`. Changeable at runtime with `log/set-level`. | | `--version` | | Print version and exit | diff --git a/services/ollama-proxy/main.go b/services/ollama-proxy/main.go index afd147dd..38ad18b4 100644 --- a/services/ollama-proxy/main.go +++ b/services/ollama-proxy/main.go @@ -34,6 +34,7 @@ func main() { ignorePersistedPort := flag.Bool("ignore-persisted-port", false, "use --port even when a persisted port exists") ipcPath := flag.String("ipc", "", "IPC endpoint: Unix domain socket path or Windows named pipe (default: stdin/stdout)") clusterDir := flag.String("cluster-dir", "", "cluster trust directory (node.crt/key + trusted pins); enables the LAN mTLS inference ingress when this node is clustered") + responseTimeout := flag.Duration("response-timeout", defaultProxyResponseTimeout, "how long to wait for a backend's response headers on a forwarded request (e.g. chat/completion) before failing over to the next candidate; raise this for large/cold-loading local models. 0 disables the timeout entirely (wait indefinitely) — only safe against a backend you trust to eventually respond or fail on its own, since a genuinely wedged backend will then hang the request forever with no automatic failover") showVersion := flag.Bool("version", false, "print version and exit") resolveLevel := applog.RegisterFlag(nil, slog.LevelInfo) flag.Parse() @@ -84,6 +85,7 @@ func main() { codec := NewCodec(transport) disc := NewDiscovery() proxy := NewProxy(codec, disc, effectivePort) + proxy.SetResponseTimeout(*responseTimeout) for _, aliasAddress := range aliasAddresses { if err := proxy.setLoopbackAlias(aliasAddress); err != nil { log.Fatalf("invalid alias address: %v", err) diff --git a/services/ollama-proxy/proxy.go b/services/ollama-proxy/proxy.go index ad4ac7a2..5242c3e1 100644 --- a/services/ollama-proxy/proxy.go +++ b/services/ollama-proxy/proxy.go @@ -382,6 +382,14 @@ type Proxy struct { plainTransport *http.Transport peerTransports map[string]*http.Transport + // responseTimeout is the ResponseHeaderTimeout applied to every forwarding + // transport built by newProxyTransport (both the shared plain transport + // and per-peer cluster mTLS transports). Defaults to + // defaultProxyResponseTimeout in NewProxy; overridden by main.go from + // --response-timeout when the operator opts in (e.g. a large local model + // whose cold-load + prefill can exceed 120s). + responseTimeout time.Duration + // nextRequestID is a monotonic counter for tagging RequestStarted / // RequestEvent pairs. Atomic add returns the new value, so request // IDs start at 1 and never collide within a single proxy lifetime. @@ -403,13 +411,78 @@ type Proxy struct { func NewProxy(codec *Codec, discovery *Discovery, port int) *Proxy { return &Proxy{ - codec: codec, - discovery: discovery, - port: port, - targets: reach.NewChooser(), - runID: newRunID(), - activity: nodeactivity.NewReporter(activityReportInterval), + codec: codec, + discovery: discovery, + port: port, + targets: reach.NewChooser(), + runID: newRunID(), + activity: nodeactivity.NewReporter(activityReportInterval), + responseTimeout: defaultProxyResponseTimeout, + } +} + +// SetResponseTimeout overrides the ResponseHeaderTimeout used by the +// forwarding transports (plain and cluster-peer) for requests dispatched +// after this call. Zero means "no timeout" — matching http.Transport's own +// ResponseHeaderTimeout semantics — and is accepted deliberately: an engine +// that only flushes once a tool call finishes composing, or a cold load on a +// large model, can legitimately take longer than any fixed bound. A negative +// duration is invalid and ignored, leaving whatever timeout is already +// configured (defaultProxyResponseTimeout unless this was already called). +// Only safe to call before Run/serveHTTP starts accepting traffic, or from +// the same goroutine that owns startup — it does not itself synchronize with +// in-flight newProxyTransport callers. +func (p *Proxy) SetResponseTimeout(d time.Duration) { + if d < 0 { + return } + p.responseTimeout = d +} + +// progressLogInterval is how often loggingRoundTripper reports that a +// forwarded request is still awaiting response headers. Independent of +// responseTimeout: it fires whether or not a timeout is even configured, so a +// long wait (or an unbounded one, with --response-timeout 0) shows up as +// periodic progress rather than silence that looks identical to a hang. +// +// A var (not a const), matching idleClientWriteTimeout above, only so a test +// can shorten it to observe a log line without waiting 30 real seconds; +// production never reassigns it. +var progressLogInterval = 30 * time.Second + +// loggingRoundTripper wraps a candidate's transport so a request in flight +// logs periodic "still waiting" progress until RoundTrip returns — either a +// response (success or a status the caller will retry/reject) or a transport +// error (dial failure, or the ResponseHeaderTimeout itself firing). It adds no +// timeout of its own; it only narrates the wait that responseTimeout (or the +// lack of one) already governs. +type loggingRoundTripper struct { + http.RoundTripper + reqID string + nodeID string + target string + path string +} + +func (t loggingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + start := time.Now() + done := make(chan struct{}) + defer close(done) + go func() { + ticker := time.NewTicker(progressLogInterval) + defer ticker.Stop() + for { + select { + case <-done: + return + case <-ticker.C: + slog.Info("proxy request still awaiting response headers", + "id", t.reqID, "node_id", t.nodeID, "target", t.target, + "path", t.path, "elapsed_ms", time.Since(start).Milliseconds()) + } + } + }() + return t.RoundTripper.RoundTrip(req) } // activityReportInterval is how often a single node's streaming may raise a @@ -526,11 +599,18 @@ func (p *Proxy) Run(ctx context.Context) error { // Timeouts for upstream connections. Logged at startup so they're always // present in any captured log for post-mortem analysis. const ( - proxyDialTimeout = 10 * time.Second - proxyKeepAlive = 30 * time.Second - proxyResponseTimeout = 120 * time.Second - proxyMaxIdleConns = 50 - proxyIdleConnTimeout = 90 * time.Second + proxyDialTimeout = 10 * time.Second + proxyKeepAlive = 30 * time.Second + // defaultProxyResponseTimeout is the ResponseHeaderTimeout applied to the + // main forwarding transport when nothing overrides it. It bounds how long + // the proxy waits for a backend (e.g. an Ollama/LM Studio node) to send + // the first response bytes for a real chat/completion request — cold model + // load, prefill, and a slow time-to-first-token all count against it. + // Configurable via --response-timeout (see main.go); this const is only + // the fallback for callers that don't set Proxy.responseTimeout. + defaultProxyResponseTimeout = 120 * time.Second + proxyMaxIdleConns = 50 + proxyIdleConnTimeout = 90 * time.Second // Inbound http.Server limits — keep IdleTimeout aligned with client // IdleConnTimeout so idle keep-alives are reaped on both sides. proxyReadHeaderTimeout = 10 * time.Second @@ -604,7 +684,7 @@ func (p *Proxy) serveHTTP(ctx context.Context, ln net.Listener) { slog.Info("proxy timeouts configured", "dial_timeout", proxyDialTimeout, "keep_alive", proxyKeepAlive, - "response_header_timeout", proxyResponseTimeout, + "response_header_timeout", p.responseTimeout, "max_idle_conns", proxyMaxIdleConns, "idle_conn_timeout", proxyIdleConnTimeout, ) @@ -867,13 +947,20 @@ func (p *Proxy) candidateTransport(c candidate) *http.Transport { return p.peerHTTPTransport(c.peerUUID) } -func newProxyTransport(tlsCfg *tls.Config) *http.Transport { +// newProxyTransport builds the main forwarding transport used to reach a +// backend Ollama/LM Studio node — the shared plain transport for a local/self +// candidate, or a per-peer transport for cluster mTLS. ResponseHeaderTimeout +// is p.responseTimeout (defaultProxyResponseTimeout unless overridden via +// SetResponseTimeout / --response-timeout), which bounds how long the proxy +// waits for the backend to start responding to a real request before failing +// over to the next candidate. +func (p *Proxy) newProxyTransport(tlsCfg *tls.Config) *http.Transport { tr := &http.Transport{ DialContext: (&net.Dialer{ Timeout: proxyDialTimeout, KeepAlive: proxyKeepAlive, }).DialContext, - ResponseHeaderTimeout: proxyResponseTimeout, + ResponseHeaderTimeout: p.responseTimeout, MaxIdleConns: proxyMaxIdleConns, MaxIdleConnsPerHost: proxyMaxIdleConns, IdleConnTimeout: proxyIdleConnTimeout, @@ -888,7 +975,7 @@ func (p *Proxy) plainHTTPTransport() *http.Transport { p.transportMu.Lock() defer p.transportMu.Unlock() if p.plainTransport == nil { - p.plainTransport = newProxyTransport(nil) + p.plainTransport = p.newProxyTransport(nil) } return p.plainTransport } @@ -904,13 +991,13 @@ func (p *Proxy) peerHTTPTransport(peerUUID string) *http.Transport { delete(p.peerTransports, peerUUID) } if p.mesh == nil { - return newProxyTransport(nil) + return p.newProxyTransport(nil) } cfg, ok := p.mesh.ClientTLSConfig(peerUUID) if !ok { - return newProxyTransport(nil) + return p.newProxyTransport(nil) } - tr := newProxyTransport(cfg) + tr := p.newProxyTransport(cfg) if p.peerTransports == nil { p.peerTransports = make(map[string]*http.Transport) } @@ -1332,7 +1419,19 @@ func (p *Proxy) handleHTTP(w http.ResponseWriter, r *http.Request) { }, // A remote cluster peer is dialed over mTLS (per-peer pinned config); // self/manual candidates use the plain transport. See candidateTransport. - Transport: p.candidateTransport(cand), + // Wrapped so a long, legitimate wait (large-context prefill, a + // tool-call composing on the other end) shows up as periodic + // progress in the logs instead of silence indistinguishable from a + // hang — especially relevant once --response-timeout is raised or + // disabled (0), where nothing else will report that the request is + // still alive until it finally succeeds or fails. + Transport: loggingRoundTripper{ + RoundTripper: p.candidateTransport(cand), + reqID: reqID, + nodeID: cand.id, + target: cand.url.Host, + path: r.URL.Path, + }, // ModifyResponse fires when the upstream's status line + headers // have arrived but before the body streams. That's both the retry // decision point and, on commit, the time-to-first-byte boundary. diff --git a/services/ollama-proxy/transport_pool_test.go b/services/ollama-proxy/transport_pool_test.go index 6c5741c4..44fba126 100644 --- a/services/ollama-proxy/transport_pool_test.go +++ b/services/ollama-proxy/transport_pool_test.go @@ -6,6 +6,7 @@ package main import ( "path/filepath" "testing" + "time" "nvpair-shared/clustertrust" "nvpair-shared/clustertrusttest" @@ -23,6 +24,47 @@ func TestCandidateTransportReusesPlainTransport(t *testing.T) { } } +func TestNewProxyDefaultsResponseTimeout(t *testing.T) { + p := testProxy(NewDiscovery(), 11435) + tr := p.candidateTransport(candidate{}) + if tr.ResponseHeaderTimeout != defaultProxyResponseTimeout { + t.Fatalf("ResponseHeaderTimeout = %v, want default %v", tr.ResponseHeaderTimeout, defaultProxyResponseTimeout) + } +} + +func TestSetResponseTimeoutOverridesTransport(t *testing.T) { + p := testProxy(NewDiscovery(), 11435) + const custom = 5 * time.Minute + p.SetResponseTimeout(custom) + + tr := p.candidateTransport(candidate{}) + if tr.ResponseHeaderTimeout != custom { + t.Fatalf("ResponseHeaderTimeout = %v, want %v", tr.ResponseHeaderTimeout, custom) + } +} + +func TestSetResponseTimeoutZeroMeansNoTimeout(t *testing.T) { + p := testProxy(NewDiscovery(), 11435) + p.SetResponseTimeout(0) + + if p.responseTimeout != 0 { + t.Fatalf("responseTimeout = %v, want 0 (no timeout)", p.responseTimeout) + } + tr := p.candidateTransport(candidate{}) + if tr.ResponseHeaderTimeout != 0 { + t.Fatalf("ResponseHeaderTimeout = %v, want 0 (no timeout)", tr.ResponseHeaderTimeout) + } +} + +func TestSetResponseTimeoutIgnoresNegative(t *testing.T) { + p := testProxy(NewDiscovery(), 11435) + p.SetResponseTimeout(-1 * time.Second) + + if p.responseTimeout != defaultProxyResponseTimeout { + t.Fatalf("responseTimeout = %v, want unchanged default %v", p.responseTimeout, defaultProxyResponseTimeout) + } +} + func TestCandidateTransportReusesPeerTransport(t *testing.T) { const peerUUID = "principal-peer" clusterDir := filepath.Join(t.TempDir(), "cluster")