From 58207f7f03c3777e68e3d258222663ba83399174 Mon Sep 17 00:00:00 2001 From: liangxu Date: Fri, 24 Apr 2026 09:44:08 +0800 Subject: [PATCH] feat(desktop): add foreground publish admission scheduler Publish is now treated as the runtime foreground channel and AI monitoring as background. A new in-memory publish-scheduler owns queue selection, per-platform active locks, and 3-6s post-completion platform cooldowns. The runtime controller tracks lease-in-flight per kind so a monitor lease no longer blocks a publish lease, derives total concurrency from hardware class, current process health, and an optional GEO_DESKTOP_MAX_TOTAL_CONCURRENCY override, and reserves a publish slot when budgeting monitor capacity. Monitor admission returns zero whenever publish has backlog or an in-flight lease, so a cooldown-blocked publish queue still prevents monitor from consuming the foreground reserve. Runtime diagnostics expose publishScheduler alongside monitorScheduler. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/main/publish-scheduler.test.ts | 53 +++ .../src/main/publish-scheduler.ts | 202 ++++++++++++ .../src/main/runtime-controller.ts | 307 ++++++++++++++---- .../src/main/runtime-snapshot.ts | 2 + 4 files changed, 494 insertions(+), 70 deletions(-) create mode 100644 apps/desktop-client/src/main/publish-scheduler.test.ts create mode 100644 apps/desktop-client/src/main/publish-scheduler.ts diff --git a/apps/desktop-client/src/main/publish-scheduler.test.ts b/apps/desktop-client/src/main/publish-scheduler.test.ts new file mode 100644 index 0000000..a18dd45 --- /dev/null +++ b/apps/desktop-client/src/main/publish-scheduler.test.ts @@ -0,0 +1,53 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { + enqueuePublishTask, + getPublishSchedulerSnapshot, + initPublishScheduler, + notePublishTaskActivated, + notePublishTaskCompleted, + selectNextPublishTask, +} from "./publish-scheduler"; + +describe("publish scheduler", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-04-24T00:00:00.000Z")); + initPublishScheduler(); + }); + + afterEach(() => { + vi.useRealTimers(); + initPublishScheduler(); + }); + + it("skips same-platform queued tasks while selecting an allowed platform", () => { + enqueuePublishTask({ taskId: "toutiao-active", platform: "toutiao", routing: "rabbitmq-primary" }); + notePublishTaskActivated("toutiao-active", "toutiao"); + enqueuePublishTask({ taskId: "toutiao-next", platform: "toutiao", routing: "rabbitmq-primary" }); + enqueuePublishTask({ taskId: "zhihu-next", platform: "zhihu", routing: "rabbitmq-primary" }); + + const selected = selectNextPublishTask({ + maxConcurrency: 2, + activeCount: 1, + }); + + expect(selected?.taskId).toBe("zhihu-next"); + expect(getPublishSchedulerSnapshot().leasingCount).toBe(1); + }); + + it("holds same-platform tasks during completion cooldown", () => { + enqueuePublishTask({ taskId: "first", platform: "toutiao", routing: "rabbitmq-primary" }); + expect(selectNextPublishTask({ maxConcurrency: 2, activeCount: 0 })?.taskId).toBe("first"); + notePublishTaskActivated("first", "toutiao"); + notePublishTaskCompleted("first", "toutiao"); + + enqueuePublishTask({ taskId: "second", platform: "toutiao", routing: "rabbitmq-primary" }); + + expect(selectNextPublishTask({ maxConcurrency: 2, activeCount: 0 })).toBeNull(); + + vi.advanceTimersByTime(6_001); + + expect(selectNextPublishTask({ maxConcurrency: 2, activeCount: 0 })?.taskId).toBe("second"); + }); +}); diff --git a/apps/desktop-client/src/main/publish-scheduler.ts b/apps/desktop-client/src/main/publish-scheduler.ts new file mode 100644 index 0000000..96ac21a --- /dev/null +++ b/apps/desktop-client/src/main/publish-scheduler.ts @@ -0,0 +1,202 @@ +export type PublishSchedulerRouting = "rabbitmq-primary" | "db-recovery"; + +type PublishTaskState = "queued" | "leasing" | "active"; + +export interface PublishSchedulerTaskRecord { + taskId: string; + platform: string; + routing: PublishSchedulerRouting; + state: PublishTaskState; + enqueuedAt: number; + updatedAt: number; + lastStartedAt: number | null; +} + +export interface PublishSchedulerSelection { + taskId: string; + platform: string; + routing: PublishSchedulerRouting; +} + +export interface PublishSchedulerSnapshot { + queueDepth: number; + queuedCount: number; + leasingCount: number; + activeCount: number; + cooldownCount: number; + tasks: PublishSchedulerTaskRecord[]; + cooldownUntil: Record; +} + +interface PublishSchedulerSelectionOptions { + now?: number; + maxConcurrency: number; + activePlatforms?: ReadonlySet; + activeCount: number; +} + +const publishPlatformCooldownMinMs = 3_000; +const publishPlatformCooldownMaxMs = 6_000; + +const tasks = new Map(); +const cooldownUntil = new Map(); + +export function initPublishScheduler(): void { + tasks.clear(); + cooldownUntil.clear(); +} + +export function enqueuePublishTask(input: { + taskId: string; + platform: string; + routing: PublishSchedulerRouting; +}): void { + const taskId = input.taskId.trim(); + const platform = input.platform.trim(); + if (!taskId || !platform) { + return; + } + + const now = Date.now(); + const existing = tasks.get(taskId); + tasks.set(taskId, { + taskId, + platform, + routing: input.routing, + state: existing?.state === "active" ? "active" : "queued", + enqueuedAt: existing?.enqueuedAt ?? now, + updatedAt: now, + lastStartedAt: existing?.lastStartedAt ?? null, + }); +} + +export function selectNextPublishTask( + options: PublishSchedulerSelectionOptions, +): PublishSchedulerSelection | null { + const now = options.now ?? Date.now(); + pruneCooldowns(now); + + if (options.activeCount >= options.maxConcurrency) { + return null; + } + + const activePlatforms = new Set(options.activePlatforms ?? []); + for (const task of tasks.values()) { + if (task.state === "active" || task.state === "leasing") { + activePlatforms.add(task.platform); + } + } + + const candidates = [...tasks.values()] + .filter((task) => task.state === "queued") + .sort((left, right) => { + if (left.enqueuedAt !== right.enqueuedAt) { + return left.enqueuedAt - right.enqueuedAt; + } + return left.updatedAt - right.updatedAt; + }); + + for (const candidate of candidates) { + if (activePlatforms.has(candidate.platform)) { + continue; + } + + const platformCooldown = cooldownUntil.get(candidate.platform) ?? 0; + if (platformCooldown > now) { + continue; + } + + candidate.state = "leasing"; + candidate.updatedAt = now; + tasks.set(candidate.taskId, candidate); + return { + taskId: candidate.taskId, + platform: candidate.platform, + routing: candidate.routing, + }; + } + + return null; +} + +export function notePublishTaskActivated(taskId: string, platform: string): void { + const normalizedTaskId = taskId.trim(); + const normalizedPlatform = platform.trim(); + if (!normalizedTaskId || !normalizedPlatform) { + return; + } + + const now = Date.now(); + const existing = tasks.get(normalizedTaskId); + tasks.set(normalizedTaskId, { + taskId: normalizedTaskId, + platform: normalizedPlatform, + routing: existing?.routing ?? "db-recovery", + state: "active", + enqueuedAt: existing?.enqueuedAt ?? now, + updatedAt: now, + lastStartedAt: now, + }); +} + +export function notePublishTaskLeaseMiss(taskId: string): void { + tasks.delete(taskId); +} + +export function notePublishTaskCompleted(taskId: string, platform?: string): void { + const existing = tasks.get(taskId); + const resolvedPlatform = (platform ?? existing?.platform ?? "").trim(); + tasks.delete(taskId); + + if (resolvedPlatform) { + cooldownUntil.set(resolvedPlatform, Date.now() + randomPublishCooldownMs()); + } +} + +export function notePublishTaskCanceled(taskId: string): void { + tasks.delete(taskId); +} + +export function getPublishSchedulerSnapshot(): PublishSchedulerSnapshot { + const now = Date.now(); + pruneCooldowns(now); + const sortedTasks = [...tasks.values()].sort((left, right) => { + if (left.state !== right.state) { + return stateRank(left.state) - stateRank(right.state); + } + return left.enqueuedAt - right.enqueuedAt; + }); + + return { + queueDepth: sortedTasks.filter((task) => task.state === "queued" || task.state === "leasing").length, + queuedCount: sortedTasks.filter((task) => task.state === "queued").length, + leasingCount: sortedTasks.filter((task) => task.state === "leasing").length, + activeCount: sortedTasks.filter((task) => task.state === "active").length, + cooldownCount: cooldownUntil.size, + tasks: sortedTasks.map((task) => ({ ...task })), + cooldownUntil: Object.fromEntries(cooldownUntil.entries()), + }; +} + +function pruneCooldowns(now: number): void { + for (const [platform, until] of cooldownUntil.entries()) { + if (until <= now) { + cooldownUntil.delete(platform); + } + } +} + +function randomPublishCooldownMs(): number { + const spread = publishPlatformCooldownMaxMs - publishPlatformCooldownMinMs; + return publishPlatformCooldownMinMs + Math.floor(Math.random() * (spread + 1)); +} + +function stateRank(state: PublishTaskState): number { + if (state === "active") { + return 0; + } + if (state === "leasing") { + return 1; + } + return 2; +} diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index 2cf8eac..7b7e1e0 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -1,4 +1,4 @@ -import { hostname } from "node:os"; +import { cpus, hostname, totalmem } from "node:os"; import { ApiClientError } from "@geo/http-client"; import type { @@ -56,6 +56,16 @@ import { noteMonitorTaskLeased, selectNextMonitorTask, } from "./monitor-scheduler"; +import { + enqueuePublishTask as enqueuePublishSchedulerTask, + getPublishSchedulerSnapshot, + initPublishScheduler, + notePublishTaskActivated, + notePublishTaskCanceled, + notePublishTaskCompleted, + notePublishTaskLeaseMiss, + selectNextPublishTask, +} from "./publish-scheduler"; import { retainHiddenPlaywrightPage, startHiddenPlaywrightReaper, @@ -126,6 +136,9 @@ const leaseExtendIntervalMs = 60_000; const maxActivityItems = 60; const monitorBusinessTimeZone = "Asia/Shanghai"; const authRequiredMonitorPlatforms = new Set(["yuanbao", "kimi", "deepseek", "doubao", "wenxin"]); +const publishReservedSlots = 1; +const minimumForegroundTotalConcurrency = 2; +const maximumRuntimeTotalConcurrency = 4; interface RuntimeTaskRecord { id: string; @@ -178,6 +191,7 @@ export interface RuntimeControllerSnapshot { interface PendingTaskRequest { taskId: string; routing: RuntimeTaskRouting; + platform?: string; source?: "desktop_task" | "monitoring_lease"; } @@ -198,14 +212,13 @@ interface RuntimeState { session: DesktopRuntimeSessionSyncRequest | null; client: DesktopClientInfo | null; running: boolean; - publishQueue: PendingTaskRequest[]; - queuedPublishTaskIds: Set; tasks: Map; accounts: DesktopAccountInfo[]; accountProfiles: Map; activity: RuntimeControllerActivity[]; activeExecutions: Map; - leaseInFlight: boolean; + leaseInFlightKinds: Set<"publish" | "monitor">; + publishFallbackBackoffUntil: number; heartbeatTimer: ReturnType | null; pullTimer: ReturnType | null; dispatchWsClient: DispatchWsClient | null; @@ -224,14 +237,13 @@ const state: RuntimeState = { session: null, client: null, running: false, - publishQueue: [], - queuedPublishTaskIds: new Set(), tasks: new Map(), accounts: [], accountProfiles: new Map(), activity: [], activeExecutions: new Map(), - leaseInFlight: false, + leaseInFlightKinds: new Set<"publish" | "monitor">(), + publishFallbackBackoffUntil: 0, heartbeatTimer: null, pullTimer: null, dispatchWsClient: null, @@ -420,6 +432,7 @@ function startRuntime(): void { initScheduler(); initMonitorScheduler(); + initPublishScheduler(); startHiddenPlaywrightReaper(); state.running = true; setSchedulerPhase("running"); @@ -438,7 +451,7 @@ function startRuntime(): void { function stopRuntime(): void { state.running = false; - state.leaseInFlight = false; + state.leaseInFlightKinds.clear(); for (const execution of state.activeExecutions.values()) { execution.abortController.abort(); } @@ -471,8 +484,7 @@ function stopRuntime(): void { } function clearRuntimeState(): void { - state.publishQueue = []; - state.queuedPublishTaskIds.clear(); + initPublishScheduler(); state.tasks.clear(); state.accounts = []; state.accountProfiles.clear(); @@ -484,6 +496,7 @@ function clearRuntimeState(): void { state.lastAccountsSyncAt = 0; state.lastServerTime = null; state.lastError = null; + state.publishFallbackBackoffUntil = 0; setSchedulerQueueDepth(localQueueDepth()); } @@ -646,9 +659,16 @@ function handleTaskEvent(event: DesktopTaskEventMessage): void { } } + if ( + event.kind === "publish" + && (event.type === "task_completed" || event.type === "task_reconciled" || event.type === "task_canceled") + ) { + notePublishTaskCanceled(event.task_id); + } + if (event.type === "task_available") { if (event.kind === "publish") { - enqueuePublishTask(event.task_id, "rabbitmq-primary"); + enqueuePublishTask(next.id, next.platform, "rabbitmq-primary"); } pumpExecutionLoop(); } @@ -1036,43 +1056,40 @@ export async function unbindRuntimeAccount(accountId: string, syncVersion: numbe await syncAccounts("manual"); } -function enqueuePublishTask(taskId: string, routing: RuntimeTaskRouting): void { - if (!taskId || state.activeExecutions.has(taskId) || state.queuedPublishTaskIds.has(taskId)) { +function enqueuePublishTask(taskId: string, platform: string, routing: RuntimeTaskRouting): void { + if (!taskId || state.activeExecutions.has(taskId)) { return; } - state.publishQueue.push({ taskId, routing }); - state.queuedPublishTaskIds.add(taskId); + enqueuePublishSchedulerTask({ taskId, platform, routing }); syncSchedulerSurface(); } -function dequeuePublishTask(): PendingTaskRequest | null { - const next = state.publishQueue.shift() ?? null; - if (next) { - state.queuedPublishTaskIds.delete(next.taskId); - } - syncSchedulerSurface(); - return next; -} - function pumpExecutionLoop(): void { - if (!state.running || state.leaseInFlight) { + if (!state.running) { return; } syncSchedulerSurface(); - if (hasActivePublishExecution()) { + if (canStartAnotherPublishExecution()) { + const selected = selectNextPublishTask({ + maxConcurrency: resolveAdaptiveTotalConcurrency(), + activePlatforms: activePublishPlatforms(), + activeCount: activePublishCount(), + }); + if (selected) { + void leaseSpecificTask(selected, "publish"); + return; + } + } + + if (shouldPullPublishFallback()) { + void pullNextTask("publish"); return; } - if (state.publishQueue.length > 0) { - if (state.activeExecutions.size === 0) { - const queuedPublish = dequeuePublishTask(); - if (queuedPublish) { - void leaseSpecificTask(queuedPublish, "publish"); - } - } + if (hasPublishBacklog()) { return; } @@ -1089,30 +1106,22 @@ function pumpExecutionLoop(): void { } } - if (state.activeExecutions.size === 0) { - // Publish task_available is push-delivered over the dispatch WebSocket. We - // only fall back to the HTTP pull when the push channel is down, otherwise - // every scheduler tick would wake the /lease endpoint for no reason. - if (!state.dispatchWsConnected) { - void pullNextTask("publish"); - } - return; - } + return; } async function leaseSpecificTask( request: PendingTaskRequest, expectedKind: "publish" | "monitor", ): Promise { - if (!canRunLive(state.session) || state.leaseInFlight) { + if (!canRunLive(state.session) || isLeaseInFlight(expectedKind)) { return; } - state.leaseInFlight = true; + state.leaseInFlightKinds.add(expectedKind); try { if (expectedKind === "monitor" && request.source === "monitoring_lease") { - state.leaseInFlight = false; + state.leaseInFlightKinds.delete(expectedKind); await executeLeasedMonitoringTask(request.taskId, request.routing); return; } @@ -1121,10 +1130,12 @@ async function leaseSpecificTask( if (!leased.task) { if (expectedKind === "monitor") { noteMonitorTaskLeaseMiss(request.taskId); + } else { + notePublishTaskLeaseMiss(request.taskId); } return; } - state.leaseInFlight = false; + state.leaseInFlightKinds.delete(expectedKind); await executeLeasedTask(leased, request.routing); } catch (error) { if (expectedKind === "monitor") { @@ -1141,6 +1152,8 @@ async function leaseSpecificTask( } else { noteMonitorTaskLeaseMiss(request.taskId); } + } else { + notePublishTaskLeaseMiss(request.taskId); } if (isApiClientError(error, 401)) { @@ -1152,10 +1165,10 @@ async function leaseSpecificTask( setSchedulerError(state.lastError); recordActivity("warn", "指定任务领取失败", `${request.taskId} 未能成功领取：${state.lastError}`); } finally { - if (state.leaseInFlight) { - state.leaseInFlight = false; + if (state.leaseInFlightKinds.has(expectedKind)) { + state.leaseInFlightKinds.delete(expectedKind); syncSchedulerSurface(); - if (state.publishQueue.length > 0 || localQueueDepth() > 0) { + if (localQueueDepth() > 0) { pumpExecutionLoop(); } } @@ -1164,11 +1177,11 @@ async function leaseSpecificTask( async function pullNextTask(kind: "publish" | "monitor"): Promise { if (kind === "monitor") { - if (!canRunLive(state.session) || state.leaseInFlight) { + if (!canRunLive(state.session) || isLeaseInFlight("monitor")) { return; } - state.leaseInFlight = true; + state.leaseInFlightKinds.add("monitor"); try { const leased = await leaseDesktopTask({ kind: "monitor" }); state.lastPullAt = Date.now(); @@ -1177,7 +1190,7 @@ async function pullNextTask(kind: "publish" | "monitor"): Promise { noteSchedulerPull(); if (leased.task) { - state.leaseInFlight = false; + state.leaseInFlightKinds.delete("monitor"); await executeLeasedTask(leased, "rabbitmq-primary"); return; } @@ -1193,7 +1206,7 @@ async function pullNextTask(kind: "publish" | "monitor"): Promise { return; } } finally { - state.leaseInFlight = false; + state.leaseInFlightKinds.delete("monitor"); syncSchedulerSurface(); } @@ -1201,11 +1214,12 @@ async function pullNextTask(kind: "publish" | "monitor"): Promise { return; } - if (!canRunLive(state.session) || state.leaseInFlight) { + if (!canRunLive(state.session) || isLeaseInFlight("publish")) { return; } - state.leaseInFlight = true; + let leasedPublishTask = false; + state.leaseInFlightKinds.add("publish"); try { const leased = await leaseDesktopTask({ kind }); @@ -1215,15 +1229,19 @@ async function pullNextTask(kind: "publish" | "monitor"): Promise { noteSchedulerPull(); if (!leased.task) { + state.publishFallbackBackoffUntil = Date.now() + pullIntervalMs; return; } - state.leaseInFlight = false; + state.publishFallbackBackoffUntil = 0; + leasedPublishTask = true; + state.leaseInFlightKinds.delete("publish"); await executeLeasedTask(leased, "db-recovery"); } catch (error) { state.lastPullAt = Date.now(); state.lastPullStatus = "failed"; state.lastError = errorMessage(error); + state.publishFallbackBackoffUntil = Date.now() + pullIntervalMs; noteTransportPull(false); setSchedulerError(state.lastError); @@ -1235,18 +1253,21 @@ async function pullNextTask(kind: "publish" | "monitor"): Promise { recordActivity("warn", "兜底拉取失败", state.lastError); } finally { - state.leaseInFlight = false; + state.leaseInFlightKinds.delete("publish"); syncSchedulerSurface(); + if (!leasedPublishTask && state.running) { + queueMicrotask(() => pumpExecutionLoop()); + } } } async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise { const leaseLimit = resolveLegacyMonitoringLeaseLimit(); - if (!canRunLive(state.session) || state.leaseInFlight || leaseLimit <= 0) { + if (!canRunLive(state.session) || isLeaseInFlight("monitor") || leaseLimit <= 0) { return; } - state.leaseInFlight = true; + state.leaseInFlightKinds.add("monitor"); try { const leased = await leaseMonitoringTasks({ @@ -1272,7 +1293,7 @@ async function pullMonitoringTasks(routing: RuntimeTaskRouting): Promise { recordActivity("warn", "监控任务拉取失败", state.lastError); } finally { - state.leaseInFlight = false; + state.leaseInFlightKinds.delete("monitor"); syncSchedulerSurface(); } } @@ -1417,6 +1438,7 @@ async function executeLeasedMonitoringTask( }); noteMonitorTaskActivated(taskId); syncSchedulerSurface(); + queueMicrotask(() => pumpExecutionLoop()); try { const execution = await executeTaskAdapter(taskRecord, abortController.signal); @@ -1971,11 +1993,14 @@ async function executeLeasedTask( interruptReason: taskRecord.kind === "monitor" ? interruptReasonFromTask(taskRecord) : null, lastSafePointAt: taskRecord.kind === "monitor" ? Date.now() : null, }); - syncSchedulerSurface(); if (task.kind === "monitor") { noteMonitorTaskLeased(task, routing); + } else { + notePublishTaskActivated(taskRecord.id, taskRecord.platform); } + syncSchedulerSurface(); + queueMicrotask(() => pumpExecutionLoop()); const extendHandle = setInterval(() => { void extendActiveLease(taskRecord.id, leased.lease_token as string); @@ -2061,6 +2086,9 @@ async function executeLeasedTask( } finally { clearInterval(extendHandle); + if (taskRecord.kind === "publish") { + notePublishTaskCompleted(taskRecord.id, taskRecord.platform); + } state.activeExecutions.delete(taskRecord.id); syncSchedulerSurface(); noteSchedulerTaskFinished(); @@ -2551,8 +2579,9 @@ function syncSchedulerSurface(): void { } function localQueueDepth(): number { - const snapshot = getMonitorSchedulerSnapshot(); - return state.publishQueue.length + snapshot.queuedCount + snapshot.leasingCount; + const publish = getPublishSchedulerSnapshot(); + const monitor = getMonitorSchedulerSnapshot(); + return publish.queueDepth + monitor.queuedCount + monitor.leasingCount; } function primaryActiveTaskId(): string | null { @@ -2561,8 +2590,26 @@ function primaryActiveTaskId(): string | null { return publish?.taskId ?? active[0]?.taskId ?? null; } -function hasActivePublishExecution(): boolean { - return [...state.activeExecutions.values()].some((item) => item.kind === "publish"); +function isLeaseInFlight(kind: "publish" | "monitor"): boolean { + return state.leaseInFlightKinds.has(kind); +} + +function activePublishCount(): number { + return [...state.activeExecutions.values()].filter((item) => item.kind === "publish").length; +} + +function activePublishPlatforms(): Set { + const platforms = new Set( + [...state.activeExecutions.values()] + .filter((item) => item.kind === "publish") + .map((item) => item.platform), + ); + for (const task of getPublishSchedulerSnapshot().tasks) { + if (task.state === "active" || task.state === "leasing") { + platforms.add(task.platform); + } + } + return platforms; } function activeMonitorCount(): number { @@ -2591,8 +2638,42 @@ function activeMonitorQuestionKeys(): Set { ); } +function hasPublishBacklog(): boolean { + const snapshot = getPublishSchedulerSnapshot(); + return snapshot.queuedCount + snapshot.leasingCount > 0; +} + +function activeAdmissionCount(): number { + const publish = getPublishSchedulerSnapshot(); + const monitor = getMonitorSchedulerSnapshot(); + return state.activeExecutions.size + + publish.leasingCount + + monitor.leasingCount + + state.leaseInFlightKinds.size; +} + +function hasTotalCapacityForNewExecution(): boolean { + return activeAdmissionCount() < resolveAdaptiveTotalConcurrency(); +} + +function canStartAnotherPublishExecution(): boolean { + const snapshot = getPublishSchedulerSnapshot(); + return snapshot.queuedCount > 0 + && !isLeaseInFlight("publish") + && hasTotalCapacityForNewExecution(); +} + +function shouldPullPublishFallback(): boolean { + return !state.dispatchWsConnected + && Date.now() >= state.publishFallbackBackoffUntil + && !isLeaseInFlight("publish") + && !hasPublishBacklog() + && activePublishCount() === 0 + && hasTotalCapacityForNewExecution(); +} + function canStartAnotherMonitorExecution(): boolean { - if (state.publishQueue.length > 0 || hasActivePublishExecution()) { + if (hasPublishBacklog() || isLeaseInFlight("publish") || isLeaseInFlight("monitor")) { return false; } const limit = resolveAdaptiveMonitorConcurrency(); @@ -2600,7 +2681,8 @@ function canStartAnotherMonitorExecution(): boolean { return false; } const scheduler = getMonitorSchedulerSnapshot(); - return activeMonitorCount() + scheduler.leasingCount < limit; + return activeMonitorCount() + scheduler.leasingCount < limit + && hasTotalCapacityForNewExecution(); } function resolveLegacyMonitoringLeaseLimit(): number { @@ -2623,6 +2705,11 @@ function resolveLegacyMonitoringLeaseLimit(): number { } function resolveAdaptiveMonitorConcurrency(): number { + const backgroundBudget = resolveMonitorBackgroundBudget(); + if (backgroundBudget <= 0) { + return 0; + } + const monitorAccountCount = state.accounts.filter((account) => isAIPlatformId(account.platform) && getProjectedAccountHealth({ @@ -2633,12 +2720,12 @@ function resolveAdaptiveMonitorConcurrency(): number { }).health === "live", ).length; if (monitorAccountCount <= 1) { - return monitorAccountCount; + return Math.min(monitorAccountCount, backgroundBudget); } const metrics = getProcessMetricsSnapshot().latestSample; if (!metrics) { - return 1; + return Math.min(1, backgroundBudget); } const totalCPUUsage = metrics.processes.reduce((sum, item) => sum + item.percentCPUUsage, 0); @@ -2652,12 +2739,92 @@ function resolveAdaptiveMonitorConcurrency(): number { || mainPrivateBytesKB >= 1_000_000 || processCount >= 40 ) { - return 1; + return Math.min(1, backgroundBudget); } + return Math.min(2, backgroundBudget); +} + +function resolveMonitorBackgroundBudget(): number { + if (hasPublishBacklog() || isLeaseInFlight("publish")) { + return 0; + } + return Math.max(0, resolveAdaptiveTotalConcurrency() - publishReservedSlots); +} + +function resolveAdaptiveTotalConcurrency(): number { + const hardwareCap = resolveHardwareTotalConcurrency(); + const runtimeCap = resolveRuntimeHealthTotalConcurrency(hardwareCap); + const featureFlagCap = resolveFeatureFlagTotalConcurrency() ?? maximumRuntimeTotalConcurrency; + return clampInteger( + Math.min(hardwareCap, runtimeCap, featureFlagCap), + featureFlagCap <= 1 ? 1 : minimumForegroundTotalConcurrency, + maximumRuntimeTotalConcurrency, + ); +} + +function resolveHardwareTotalConcurrency(): number { + const cpuCount = Math.max(1, cpus().length); + const memoryGB = totalmem() / 1024 / 1024 / 1024; + + if (cpuCount >= 10 && memoryGB >= 32) { + return 4; + } + if (cpuCount >= 6 && memoryGB >= 16) { + return 3; + } return 2; } +function resolveRuntimeHealthTotalConcurrency(hardwareCap: number): number { + const metrics = getProcessMetricsSnapshot().latestSample; + if (!metrics) { + return Math.min(hardwareCap, minimumForegroundTotalConcurrency); + } + + const totalCPUUsage = metrics.processes.reduce((sum, item) => sum + item.percentCPUUsage, 0); + const totalPrivateBytesKB = metrics.totals.privateBytes; + const processCount = metrics.totals.processCount; + const mainPrivateBytesKB = metrics.mainProcess.memory?.private ?? 0; + + if ( + totalCPUUsage >= 180 + || totalPrivateBytesKB >= 2_500_000 + || mainPrivateBytesKB >= 1_000_000 + || processCount >= 40 + ) { + return minimumForegroundTotalConcurrency; + } + + if ( + totalCPUUsage >= 120 + || totalPrivateBytesKB >= 1_800_000 + || mainPrivateBytesKB >= 750_000 + || processCount >= 32 + ) { + return Math.min(hardwareCap, 3); + } + + return hardwareCap; +} + +function resolveFeatureFlagTotalConcurrency(): number | null { + const raw = process.env.GEO_DESKTOP_MAX_TOTAL_CONCURRENCY?.trim(); + if (!raw) { + return null; + } + + const parsed = Number.parseInt(raw, 10); + if (!Number.isFinite(parsed)) { + return null; + } + return clampInteger(parsed, 1, maximumRuntimeTotalConcurrency); +} + +function clampInteger(value: number, min: number, max: number): number { + return Math.min(max, Math.max(min, Math.floor(value))); +} + function handleAuthExpired(error: unknown): void { const message = errorMessage(error); state.lastError = message; diff --git a/apps/desktop-client/src/main/runtime-snapshot.ts b/apps/desktop-client/src/main/runtime-snapshot.ts index 9b52330..d3994ee 100644 --- a/apps/desktop-client/src/main/runtime-snapshot.ts +++ b/apps/desktop-client/src/main/runtime-snapshot.ts @@ -7,6 +7,7 @@ import { collectRecoverySnapshot } from "./crash-recovery"; import { getKeepAlivePlan } from "./keep-alive"; import { getLeaseManagerSnapshot } from "./lease-manager"; import { getMonitorSchedulerSnapshot } from "./monitor-scheduler"; +import { getPublishSchedulerSnapshot } from "./publish-scheduler"; import { getObservedRequestSnapshot } from "./network-observer"; import { getHiddenPlaywrightSnapshot } from "./playwright-cdp"; import { getRateLimiterSnapshot } from "./rate-limiter"; @@ -169,6 +170,7 @@ function createLiveRuntimeSnapshot(controller: ReturnType