mirror of
https://github.com/stablyai/orca.git
synced 2026-10-19 16:03:26 +00:00
fix(runtime): preserve pacing and release yield callbacks
Adversarial review found that concurrent producers could retain resolved callbacks until global quiescence. Route renderer yields by token and delete each resolver before resuming its producer. Keep timer pacing in terminal paste and accepted-write loops where SSH and local PTYs do not provide drain acknowledgement. Use the shared scheduler for the OpenCode scanner.
This commit is contained in:
@@ -3,6 +3,7 @@ import { existsSync } from 'node:fs'
|
||||
import { readdir, realpath, stat } from 'node:fs/promises'
|
||||
import { homedir } from 'node:os'
|
||||
import { basename, isAbsolute, join, posix, win32 } from 'node:path'
|
||||
import { yieldToEventLoop } from '../../shared/event-loop-yield'
|
||||
import type { Repo } from '../../shared/types'
|
||||
import { areWorktreePathsEqual } from '../ipc/worktree-logic'
|
||||
import Database from '../sqlite/sync-database'
|
||||
@@ -144,10 +145,6 @@ export async function getProcessedDatabaseInfo(
|
||||
}
|
||||
}
|
||||
|
||||
async function yieldToEventLoop(): Promise<void> {
|
||||
await new Promise((resolve) => setImmediate(resolve))
|
||||
}
|
||||
|
||||
function getProjectJoin(db: Database.Database): string {
|
||||
return tableExists(db, 'project') && columnExists(db, 'session', 'project_id')
|
||||
? 'LEFT JOIN project p ON p.id = s.project_id'
|
||||
|
||||
@@ -10,7 +10,6 @@ import {
|
||||
isTerminalInputTooLargeWithDeferredMeasurement,
|
||||
iterateTerminalInputChunks
|
||||
} from '../../../../shared/terminal-input'
|
||||
import { yieldToEventLoop } from '../../../../shared/event-loop-yield'
|
||||
import { isRuntimeOwnedSshTargetId } from '../../../../shared/execution-host'
|
||||
import {
|
||||
ptyDataHandlers,
|
||||
@@ -623,6 +622,10 @@ export function createIpcPtyTransport(opts: IpcPtyTransportOptions = {}): PtyTra
|
||||
commit: clearAccumulatedState
|
||||
}
|
||||
|
||||
function yieldToInputWriteDrain(): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, 0))
|
||||
}
|
||||
|
||||
async function writeAcceptedPtyInput(id: string, data: string): Promise<boolean> {
|
||||
try {
|
||||
const tooLarge = isTerminalInputTooLargeWithDeferredMeasurement(data)
|
||||
@@ -641,7 +644,7 @@ export function createIpcPtyTransport(opts: IpcPtyTransportOptions = {}): PtyTra
|
||||
}
|
||||
chunk = chunks.next()
|
||||
if (!chunk.done) {
|
||||
await yieldToEventLoop()
|
||||
await yieldToInputWriteDrain()
|
||||
}
|
||||
}
|
||||
return true
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
import { yieldToEventLoop } from '../../../../shared/event-loop-yield'
|
||||
import { BRACKETED_PASTE_END, BRACKETED_PASTE_START } from './terminal-bracketed-paste'
|
||||
import { iterateTerminalPastePlanChunks } from './terminal-paste-chunks'
|
||||
import { createRedactedPasteExecutionDiagnostic } from './terminal-paste-diagnostics'
|
||||
@@ -31,7 +30,7 @@ export async function executeTerminalPastePlan(
|
||||
writePty,
|
||||
isTargetCurrent,
|
||||
canContinue,
|
||||
yieldToEventLoop: yieldBetweenChunks = yieldToEventLoop,
|
||||
yieldToEventLoop = defaultYieldToEventLoop,
|
||||
operationTimeoutMs = getTerminalPasteOperationTimeoutMs(plan),
|
||||
now = defaultNow
|
||||
}: ExecuteTerminalPastePlanArgs
|
||||
@@ -102,7 +101,7 @@ export async function executeTerminalPastePlan(
|
||||
} else if (chunk === BRACKETED_PASTE_END) {
|
||||
bracketedPasteOpen = false
|
||||
}
|
||||
await yieldBetweenChunks()
|
||||
await yieldToEventLoop()
|
||||
}
|
||||
return finish('pasted', chunksWritten)
|
||||
}
|
||||
@@ -141,3 +140,7 @@ function result(
|
||||
function defaultNow(): number {
|
||||
return globalThis.performance?.now?.() ?? Date.now()
|
||||
}
|
||||
|
||||
function defaultYieldToEventLoop(): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, 0))
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { yieldToEventLoop } from './event-loop-yield'
|
||||
import { getPendingRendererYieldCountForTesting, yieldToEventLoop } from './event-loop-yield'
|
||||
|
||||
afterEach(() => {
|
||||
vi.unstubAllEnvs()
|
||||
@@ -18,8 +18,9 @@ describe('yieldToEventLoop', () => {
|
||||
expect(scheduleImmediate).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('posts one macrotask per renderer yield', async () => {
|
||||
it('releases callbacks during sustained concurrent renderer yields', async () => {
|
||||
const postMessage = vi.fn()
|
||||
let peakPendingAfterResolution = 0
|
||||
vi.stubEnv('VITEST', 'false')
|
||||
vi.stubGlobal('window', {})
|
||||
vi.stubGlobal(
|
||||
@@ -29,14 +30,25 @@ describe('yieldToEventLoop', () => {
|
||||
port2 = {
|
||||
postMessage: (data: unknown): void => {
|
||||
postMessage(data)
|
||||
queueMicrotask(() => this.port1.onmessage?.({ data } as MessageEvent))
|
||||
setTimeout(() => this.port1.onmessage?.({ data } as MessageEvent), 0)
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
await Promise.all([yieldToEventLoop(), yieldToEventLoop()])
|
||||
const runProducer = async (): Promise<void> => {
|
||||
for (let index = 0; index < 20; index += 1) {
|
||||
await yieldToEventLoop()
|
||||
peakPendingAfterResolution = Math.max(
|
||||
peakPendingAfterResolution,
|
||||
getPendingRendererYieldCountForTesting()
|
||||
)
|
||||
}
|
||||
}
|
||||
await Promise.all([runProducer(), runProducer()])
|
||||
|
||||
expect(postMessage).toHaveBeenCalledTimes(2)
|
||||
expect(postMessage).toHaveBeenCalledTimes(40)
|
||||
expect(peakPendingAfterResolution).toBeLessThanOrEqual(1)
|
||||
expect(getPendingRendererYieldCountForTesting()).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -2,8 +2,8 @@ type ImmediateGlobal = typeof globalThis & {
|
||||
setImmediate?: (callback: () => void) => unknown
|
||||
}
|
||||
|
||||
const pendingRendererYields: (() => void)[] = []
|
||||
let nextRendererYieldIndex = 0
|
||||
const pendingRendererYields = new Map<number, () => void>()
|
||||
let nextRendererYieldId = 0
|
||||
let rendererYieldChannel: MessageChannel | null = null
|
||||
|
||||
function isVitestEnvironment(): boolean {
|
||||
@@ -13,22 +13,25 @@ function isVitestEnvironment(): boolean {
|
||||
function getRendererYieldChannel(): MessageChannel {
|
||||
if (!rendererYieldChannel) {
|
||||
rendererYieldChannel = new globalThis.MessageChannel()
|
||||
rendererYieldChannel.port1.onmessage = () => {
|
||||
const resolve = pendingRendererYields[nextRendererYieldIndex]
|
||||
rendererYieldChannel.port1.onmessage = (event) => {
|
||||
const yieldId = event.data
|
||||
const resolve = typeof yieldId === 'number' ? pendingRendererYields.get(yieldId) : undefined
|
||||
if (!resolve) {
|
||||
return
|
||||
}
|
||||
nextRendererYieldIndex += 1
|
||||
if (nextRendererYieldIndex === pendingRendererYields.length) {
|
||||
pendingRendererYields.length = 0
|
||||
nextRendererYieldIndex = 0
|
||||
}
|
||||
pendingRendererYields.delete(yieldId)
|
||||
resolve()
|
||||
}
|
||||
}
|
||||
return rendererYieldChannel
|
||||
}
|
||||
|
||||
/** @internal */
|
||||
export function getPendingRendererYieldCountForTesting(): number {
|
||||
return pendingRendererYields.size
|
||||
}
|
||||
|
||||
/** Yields to another runnable task without a timer clamp when supported. */
|
||||
export function yieldToEventLoop(): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
// Vitest fake timers cannot advance MessageChannel tasks.
|
||||
@@ -45,8 +48,10 @@ export function yieldToEventLoop(): Promise<void> {
|
||||
|
||||
if (typeof globalThis.MessageChannel === 'function') {
|
||||
// Posted tasks avoid Chromium's nested-timer clamp while still yielding to input and paint.
|
||||
pendingRendererYields.push(resolve)
|
||||
getRendererYieldChannel().port2.postMessage(undefined)
|
||||
const yieldId = nextRendererYieldId
|
||||
nextRendererYieldId += 1
|
||||
pendingRendererYields.set(yieldId, resolve)
|
||||
getRendererYieldChannel().port2.postMessage(yieldId)
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user