root/geo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(desktop): add foreground publish admission scheduler

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
liangxu
2026-04-24 09:44:08 +08:00
parent af600a6910
commit 58207f7f03
4 changed files with 494 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/desktop-client/src/main/runtime-controller.ts
+237 -70
View File
@@ -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<string>;
tasks: Map<string, RuntimeTaskRecord>;
accounts: DesktopAccountInfo[];
accountProfiles: Map<string, PublishAccountProfile>;
activity: RuntimeControllerActivity[];
activeExecutions: Map<string, ActiveRuntimeExecution>;
leaseInFlight: boolean;
leaseInFlightKinds: Set<"publish" | "monitor">;
publishFallbackBackoffUntil: number;
heartbeatTimer: ReturnType<typeof setInterval> | null;
pullTimer: ReturnType<typeof setInterval> | null;
dispatchWsClient: DispatchWsClient | null;
@@ -224,14 +237,13 @@ const state: RuntimeState = {
session: null,
client: null,
running: false,
publishQueue: [],
queuedPublishTaskIds: new Set<string>(),
tasks: new Map<string, RuntimeTaskRecord>(),
accounts: [],
accountProfiles: new Map<string, PublishAccountProfile>(),
activity: [],
activeExecutions: new Map<string, ActiveRuntimeExecution>(),
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<void> {
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<void> {
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<void> {
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<void> {
return;
}
} finally {
state.leaseInFlight = false;
state.leaseInFlightKinds.delete("monitor");
syncSchedulerSurface();
}
@@ -1201,11 +1214,12 @@ async function pullNextTask(kind: "publish" | "monitor"): Promise<void> {
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<void> {
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<void> {
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<void> {
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<void> {
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<string> {
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<string> {
);
}
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;