mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
fix(git): close final admission gaps
This commit is contained in:
@@ -14,19 +14,17 @@ type WaiterLane = {
|
||||
|
||||
const TIERS = ['interactive', 'status', 'background'] as const
|
||||
const createLane = (): WaiterLane => ({ items: [], head: 0, count: 0 })
|
||||
type TierLanes = Record<GitAdmissionTier, Map<string | null, WaiterLane>>
|
||||
const createTierLanes = (): TierLanes => ({
|
||||
interactive: new Map(),
|
||||
status: new Map(),
|
||||
background: new Map()
|
||||
})
|
||||
|
||||
export class GitAdmissionWaiterQueue {
|
||||
private readonly lanes: Record<AdmissionClass, Record<GitAdmissionTier, WaiterLane>> = {
|
||||
general: {
|
||||
interactive: createLane(),
|
||||
status: createLane(),
|
||||
background: createLane()
|
||||
},
|
||||
network: {
|
||||
interactive: createLane(),
|
||||
status: createLane(),
|
||||
background: createLane()
|
||||
}
|
||||
private readonly lanes: Record<AdmissionClass, TierLanes> = {
|
||||
general: createTierLanes(),
|
||||
network: createTierLanes()
|
||||
}
|
||||
private readonly countsByBudget = new Map<string, number>()
|
||||
private totalCount = 0
|
||||
@@ -36,7 +34,12 @@ export class GitAdmissionWaiterQueue {
|
||||
}
|
||||
|
||||
enqueue(waiter: AdmissionWaiter): void {
|
||||
const lane = this.lanes[waiter.admissionClass][waiter.tier]
|
||||
const lanes = this.lanes[waiter.admissionClass][waiter.tier]
|
||||
let lane = lanes.get(waiter.route)
|
||||
if (!lane) {
|
||||
lane = createLane()
|
||||
lanes.set(waiter.route, lane)
|
||||
}
|
||||
lane.items.push(waiter)
|
||||
lane.count += 1
|
||||
this.totalCount += 1
|
||||
@@ -46,7 +49,11 @@ export class GitAdmissionWaiterQueue {
|
||||
}
|
||||
|
||||
dequeue(waiter: AdmissionWaiter): void {
|
||||
const lane = this.lanes[waiter.admissionClass][waiter.tier]
|
||||
const lanes = this.lanes[waiter.admissionClass][waiter.tier]
|
||||
const lane = lanes.get(waiter.route)
|
||||
if (!lane) {
|
||||
return
|
||||
}
|
||||
lane.count -= 1
|
||||
this.totalCount -= 1
|
||||
for (const key of waiter.budgetKeys) {
|
||||
@@ -58,6 +65,9 @@ export class GitAdmissionWaiterQueue {
|
||||
}
|
||||
}
|
||||
this.compact(lane)
|
||||
if (lane.count === 0) {
|
||||
lanes.delete(waiter.route)
|
||||
}
|
||||
}
|
||||
|
||||
hasBudget(key: string): boolean {
|
||||
@@ -68,8 +78,9 @@ export class GitAdmissionWaiterQueue {
|
||||
return (['general', 'network'] as const)
|
||||
.flatMap((admissionClass) =>
|
||||
TIERS.flatMap((tier) => {
|
||||
const lane = this.lanes[admissionClass][tier]
|
||||
return lane.items.slice(lane.head).filter((waiter) => waiter.state === 'queued')
|
||||
return [...this.lanes[admissionClass][tier].values()].flatMap((lane) =>
|
||||
lane.items.slice(lane.head).filter((waiter) => waiter.state === 'queued')
|
||||
)
|
||||
})
|
||||
)
|
||||
.sort((left, right) => left.id - right.id)
|
||||
@@ -83,29 +94,31 @@ export class GitAdmissionWaiterQueue {
|
||||
): SelectedWaiter | null {
|
||||
let selected: (SelectedWaiter & { tier: number }) | null = null
|
||||
for (const rawTier of TIERS) {
|
||||
const lane = this.lanes[admissionClass][rawTier]
|
||||
for (let index = lane.head; index < lane.items.length; index += 1) {
|
||||
const waiter = lane.items[index]
|
||||
if (waiter.state !== 'queued') {
|
||||
continue
|
||||
for (const lane of this.lanes[admissionClass][rawTier].values()) {
|
||||
for (let index = lane.head; index < lane.items.length; index += 1) {
|
||||
const waiter = lane.items[index]
|
||||
if (waiter.state !== 'queued') {
|
||||
continue
|
||||
}
|
||||
if (waiter.signal?.aborted) {
|
||||
abort(waiter)
|
||||
continue
|
||||
}
|
||||
const slotKind = slotKindFor(waiter)
|
||||
if (slotKind) {
|
||||
const tier = effectiveTier(waiter)
|
||||
if (
|
||||
!selected ||
|
||||
tier < selected.tier ||
|
||||
(tier === selected.tier && waiter.id < selected.waiter.id)
|
||||
) {
|
||||
selected = { waiter, slotKind, tier }
|
||||
}
|
||||
}
|
||||
// Every waiter in this route lane has the same budget shape, so a
|
||||
// blocked head proves the rest cannot fit either.
|
||||
break
|
||||
}
|
||||
if (waiter.signal?.aborted) {
|
||||
abort(waiter)
|
||||
continue
|
||||
}
|
||||
const slotKind = slotKindFor(waiter)
|
||||
if (!slotKind) {
|
||||
continue
|
||||
}
|
||||
const tier = effectiveTier(waiter)
|
||||
if (
|
||||
!selected ||
|
||||
tier < selected.tier ||
|
||||
(tier === selected.tier && waiter.id < selected.waiter.id)
|
||||
) {
|
||||
selected = { waiter, slotKind, tier }
|
||||
}
|
||||
break
|
||||
}
|
||||
}
|
||||
return selected
|
||||
|
||||
@@ -4,6 +4,7 @@ import { execFileCapture } from './exec-file-capture'
|
||||
import { resolveGitCommand } from './git-command-resolution'
|
||||
import { DEFAULT_GIT_MAX_BUFFER, type GitExecOptions } from './git-exec-options'
|
||||
import { promptGuardGitEnv } from './git-process-env'
|
||||
import { acquireGitAdmission } from './git-subprocess-admission'
|
||||
|
||||
export type GitSshPolicyMode =
|
||||
| 'default'
|
||||
@@ -131,6 +132,18 @@ export async function buildNetworkSshPolicyEnv(options: GitExecOptions): Promise
|
||||
// Why fenced: a login-shell banner here reads as a user-configured sshCommand,
|
||||
// which skips the BatchMode fallback below and disarms the no-prompt guard.
|
||||
const resolved = resolveGitCommand(['config', '--get', 'core.sshCommand'], options, true, true)
|
||||
const probeArgs = ['config', '--get', 'core.sshCommand']
|
||||
const grant = await acquireGitAdmission({
|
||||
args: probeArgs,
|
||||
cwd: options.cwd,
|
||||
wslDistro: options.wslDistro,
|
||||
tier: options.admissionTier,
|
||||
signal: options.signal
|
||||
})
|
||||
let reportTerminated: () => void = () => {}
|
||||
const terminated = new Promise<void>((resolve) => {
|
||||
reportTerminated = resolve
|
||||
})
|
||||
let configuredCommand = ''
|
||||
try {
|
||||
const { stdout } = await execFileCapture(resolved.binary, resolved.args, {
|
||||
@@ -139,12 +152,15 @@ export async function buildNetworkSshPolicyEnv(options: GitExecOptions): Promise
|
||||
maxBuffer: DEFAULT_GIT_MAX_BUFFER,
|
||||
timeout: CORE_SSH_COMMAND_PROBE_TIMEOUT_MS,
|
||||
env: promptEnv,
|
||||
signal: options.signal
|
||||
signal: options.signal,
|
||||
onChildTerminated: reportTerminated
|
||||
})
|
||||
const payload = resolved.captured?.readStdout(String(stdout)) ?? String(stdout)
|
||||
configuredCommand = payload.trim()
|
||||
} catch {
|
||||
configuredCommand = ''
|
||||
} finally {
|
||||
void terminated.then(grant.release)
|
||||
}
|
||||
|
||||
if (!configuredCommand) {
|
||||
|
||||
@@ -369,6 +369,48 @@ describe('GitAdmissionScheduler', () => {
|
||||
expect(scheduler.snapshot()).toMatchObject({ queued: 0, queuedWaiters: [] })
|
||||
})
|
||||
|
||||
it('skips a 4k saturated-route lane without rescanning its blocked prefix', async () => {
|
||||
const scheduler = new GitAdmissionScheduler({
|
||||
generalCap: 2,
|
||||
generalHeadroom: 0,
|
||||
routeCap: 1,
|
||||
routeHeadroom: 0
|
||||
})
|
||||
const routedRunning = await scheduler.acquire({
|
||||
...local('background'),
|
||||
wslDistro: 'Ubuntu'
|
||||
})
|
||||
const localRunning = await scheduler.acquire(local('background'))
|
||||
let abortedReads = 0
|
||||
const signal = {
|
||||
get aborted() {
|
||||
abortedReads += 1
|
||||
return false
|
||||
},
|
||||
addEventListener: () => {},
|
||||
removeEventListener: () => {}
|
||||
} as unknown as AbortSignal
|
||||
const count = 4_000
|
||||
const routed = Array.from({ length: count }, () =>
|
||||
scheduler
|
||||
.acquire({ ...local('background'), wslDistro: 'Ubuntu', signal })
|
||||
.then((grant) => grant.release())
|
||||
)
|
||||
const localWork = Array.from({ length: count }, () =>
|
||||
scheduler.acquire({ ...local('background'), signal }).then((grant) => grant.release())
|
||||
)
|
||||
abortedReads = 0
|
||||
|
||||
localRunning.release()
|
||||
await Promise.all(localWork)
|
||||
expect(scheduler.snapshot().queued).toBe(count)
|
||||
|
||||
routedRunning.release()
|
||||
await Promise.all(routed)
|
||||
expect(abortedReads).toBeLessThan(30_000)
|
||||
expect(scheduler.snapshot()).toMatchObject({ queued: 0, queuedWaiters: [] })
|
||||
})
|
||||
|
||||
it('captures killswitch state in each release closure', async () => {
|
||||
const scheduler = schedulerWithOneSlot()
|
||||
_resetGitAdmissionForTests(scheduler)
|
||||
|
||||
Reference in New Issue
Block a user