mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
fix(git): harden admission across runtime paths
This commit is contained in:
@@ -0,0 +1,85 @@
|
||||
import type { AdmissionWaiter } from './git-admission-state'
|
||||
|
||||
export type WaiterLane = {
|
||||
items: AdmissionWaiter[]
|
||||
head: number
|
||||
count: number
|
||||
baseEligible: boolean
|
||||
headroomEligible: boolean
|
||||
version: number
|
||||
}
|
||||
|
||||
export type Candidate = { lane: WaiterLane; waiter: AdmissionWaiter; version: number }
|
||||
|
||||
const CANDIDATE_HEAP_COMPACTION_SLACK = 64
|
||||
|
||||
export class CandidateHeap {
|
||||
private items: Candidate[] = []
|
||||
|
||||
get size(): number {
|
||||
return this.items.length
|
||||
}
|
||||
|
||||
push(candidate: Candidate): void {
|
||||
this.items.push(candidate)
|
||||
let index = this.items.length - 1
|
||||
while (index > 0) {
|
||||
const parent = Math.floor((index - 1) / 2)
|
||||
if (this.items[parent].waiter.id <= candidate.waiter.id) {
|
||||
break
|
||||
}
|
||||
this.items[index] = this.items[parent]
|
||||
index = parent
|
||||
}
|
||||
this.items[index] = candidate
|
||||
}
|
||||
|
||||
peek(valid: (candidate: Candidate) => boolean): Candidate | null {
|
||||
// Eligibility changes invalidate by version so route updates stay O(log N).
|
||||
while (this.items.length > 0 && !valid(this.items[0])) {
|
||||
this.pop()
|
||||
}
|
||||
return this.items[0] ?? null
|
||||
}
|
||||
|
||||
compactIfOversized(maxLiveCandidates: number, valid: (candidate: Candidate) => boolean): void {
|
||||
if (this.items.length <= maxLiveCandidates * 2 + CANDIDATE_HEAP_COMPACTION_SLACK) {
|
||||
return
|
||||
}
|
||||
this.items = this.items.filter(valid)
|
||||
for (let index = Math.floor(this.items.length / 2) - 1; index >= 0; index -= 1) {
|
||||
this.siftDown(index)
|
||||
}
|
||||
}
|
||||
|
||||
private pop(): void {
|
||||
const last = this.items.pop()
|
||||
if (!last || this.items.length === 0) {
|
||||
return
|
||||
}
|
||||
this.items[0] = last
|
||||
this.siftDown(0)
|
||||
}
|
||||
|
||||
private siftDown(index: number): void {
|
||||
const candidate = this.items[index]
|
||||
let cursor = index
|
||||
while (true) {
|
||||
const left = cursor * 2 + 1
|
||||
if (left >= this.items.length) {
|
||||
break
|
||||
}
|
||||
const right = left + 1
|
||||
const child =
|
||||
right < this.items.length && this.items[right].waiter.id < this.items[left].waiter.id
|
||||
? right
|
||||
: left
|
||||
if (this.items[child].waiter.id >= candidate.waiter.id) {
|
||||
break
|
||||
}
|
||||
this.items[cursor] = this.items[child]
|
||||
cursor = child
|
||||
}
|
||||
this.items[cursor] = candidate
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import type { AdmissionClass, AdmissionSlotKind, AdmissionWaiter } from './git-admission-state'
|
||||
import { CandidateHeap, type Candidate, type WaiterLane } from './git-admission-candidate-heap'
|
||||
import type { GitAdmissionTier } from './git-exec-options'
|
||||
|
||||
type SelectedWaiter = {
|
||||
@@ -6,15 +7,6 @@ type SelectedWaiter = {
|
||||
slotKind: AdmissionSlotKind
|
||||
}
|
||||
|
||||
type WaiterLane = {
|
||||
items: AdmissionWaiter[]
|
||||
head: number
|
||||
count: number
|
||||
baseEligible: boolean
|
||||
headroomEligible: boolean
|
||||
version: number
|
||||
}
|
||||
|
||||
const TIERS = ['interactive', 'status', 'background'] as const
|
||||
const createLane = (): WaiterLane => ({
|
||||
items: [],
|
||||
@@ -30,59 +22,6 @@ const createTierLanes = (): TierLanes => ({
|
||||
status: new Map(),
|
||||
background: new Map()
|
||||
})
|
||||
type Candidate = { lane: WaiterLane; waiter: AdmissionWaiter; version: number }
|
||||
|
||||
class CandidateHeap {
|
||||
private readonly items: Candidate[] = []
|
||||
|
||||
push(candidate: Candidate): void {
|
||||
this.items.push(candidate)
|
||||
let index = this.items.length - 1
|
||||
while (index > 0) {
|
||||
const parent = Math.floor((index - 1) / 2)
|
||||
if (this.items[parent].waiter.id <= candidate.waiter.id) {
|
||||
break
|
||||
}
|
||||
this.items[index] = this.items[parent]
|
||||
index = parent
|
||||
}
|
||||
this.items[index] = candidate
|
||||
}
|
||||
|
||||
peek(valid: (candidate: Candidate) => boolean): Candidate | null {
|
||||
// Eligibility changes invalidate by version so route updates stay O(log N).
|
||||
while (this.items.length > 0 && !valid(this.items[0])) {
|
||||
this.pop()
|
||||
}
|
||||
return this.items[0] ?? null
|
||||
}
|
||||
|
||||
private pop(): void {
|
||||
const last = this.items.pop()
|
||||
if (!last || this.items.length === 0) {
|
||||
return
|
||||
}
|
||||
let index = 0
|
||||
while (true) {
|
||||
const left = index * 2 + 1
|
||||
if (left >= this.items.length) {
|
||||
break
|
||||
}
|
||||
const right = left + 1
|
||||
const child =
|
||||
right < this.items.length && this.items[right].waiter.id < this.items[left].waiter.id
|
||||
? right
|
||||
: left
|
||||
if (this.items[child].waiter.id >= last.waiter.id) {
|
||||
break
|
||||
}
|
||||
this.items[index] = this.items[child]
|
||||
index = child
|
||||
}
|
||||
this.items[index] = last
|
||||
}
|
||||
}
|
||||
|
||||
export class GitAdmissionWaiterQueue {
|
||||
private readonly lanes: Record<AdmissionClass, TierLanes> = {
|
||||
general: createTierLanes(),
|
||||
@@ -112,6 +51,17 @@ export class GitAdmissionWaiterQueue {
|
||||
return this.totalCount
|
||||
}
|
||||
|
||||
/** @internal - exposed for bounded-storage regression tests only. */
|
||||
get candidateCountForTests(): number {
|
||||
let count = this.headroomCandidates.general.size + this.headroomCandidates.network.size
|
||||
for (const admissionClass of ['general', 'network'] as const) {
|
||||
for (const tier of TIERS) {
|
||||
count += this.baseCandidates[admissionClass][tier].size
|
||||
}
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
enqueue(waiter: AdmissionWaiter): void {
|
||||
const lanes = this.lanes[waiter.admissionClass][waiter.tier]
|
||||
let lane = lanes.get(waiter.route)
|
||||
@@ -151,9 +101,11 @@ export class GitAdmissionWaiterQueue {
|
||||
if (lane.count === 0) {
|
||||
lane.version += 1
|
||||
lanes.delete(waiter.route)
|
||||
this.compactCandidateHeaps(waiter.admissionClass, waiter.tier)
|
||||
} else if (lane.items[lane.head] !== previousHead) {
|
||||
lane.version += 1
|
||||
this.publishLaneHead(waiter.admissionClass, waiter.tier, lane)
|
||||
this.compactCandidateHeaps(waiter.admissionClass, waiter.tier)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -175,6 +127,7 @@ export class GitAdmissionWaiterQueue {
|
||||
lane.headroomEligible = headroomEligible
|
||||
lane.version += 1
|
||||
this.publishLaneHead(admissionClass, tier, lane)
|
||||
this.compactCandidateHeaps(admissionClass, tier)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -250,15 +203,31 @@ export class GitAdmissionWaiterQueue {
|
||||
}
|
||||
|
||||
private peekValid(heap: CandidateHeap, slotKind: AdmissionSlotKind): Candidate | null {
|
||||
return heap.peek(
|
||||
({ lane, waiter, version }) =>
|
||||
version === lane.version &&
|
||||
waiter === lane.items[lane.head] &&
|
||||
waiter.state === 'queued' &&
|
||||
(slotKind === 'base' ? lane.baseEligible : lane.headroomEligible)
|
||||
return heap.peek((candidate) => this.candidateIsValid(candidate, slotKind))
|
||||
}
|
||||
|
||||
private candidateIsValid(candidate: Candidate, slotKind: AdmissionSlotKind): boolean {
|
||||
const { lane, waiter, version } = candidate
|
||||
return (
|
||||
version === lane.version &&
|
||||
waiter === lane.items[lane.head] &&
|
||||
waiter.state === 'queued' &&
|
||||
(slotKind === 'base' ? lane.baseEligible : lane.headroomEligible)
|
||||
)
|
||||
}
|
||||
|
||||
private compactCandidateHeaps(admissionClass: AdmissionClass, tier: GitAdmissionTier): void {
|
||||
const liveLaneCount = this.lanes[admissionClass][tier].size
|
||||
this.baseCandidates[admissionClass][tier].compactIfOversized(liveLaneCount, (candidate) =>
|
||||
this.candidateIsValid(candidate, 'base')
|
||||
)
|
||||
if (tier === 'interactive') {
|
||||
this.headroomCandidates[admissionClass].compactIfOversized(liveLaneCount, (candidate) =>
|
||||
this.candidateIsValid(candidate, 'headroom')
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
private compact(lane: WaiterLane): void {
|
||||
let head = lane.head
|
||||
while (head < lane.items.length && lane.items[head].state !== 'queued') {
|
||||
|
||||
@@ -110,6 +110,7 @@ async function gitExecFileAsyncUnlocked(
|
||||
result = await capture(resolved)
|
||||
} catch (error) {
|
||||
if (directWslGitExitCode(error, resolved) !== null && !options.signal?.aborted) {
|
||||
await terminationState.current
|
||||
const wasMissing = invalidateMissingDirectWslGit(error, resolved)
|
||||
const fallback = resolveGitCommand(
|
||||
args,
|
||||
|
||||
@@ -285,6 +285,28 @@ describe('GitAdmissionScheduler', () => {
|
||||
running.release()
|
||||
})
|
||||
|
||||
it('bounds canceled candidate storage while the only permit stays saturated', async () => {
|
||||
const scheduler = schedulerWithOneSlot()
|
||||
const running = await scheduler.acquire(local())
|
||||
const canceled: Promise<void>[] = []
|
||||
|
||||
for (let index = 0; index < 4_000; index += 1) {
|
||||
const controller = new AbortController()
|
||||
const request = scheduler.acquire({ ...local(), signal: controller.signal }).then(
|
||||
() => undefined,
|
||||
(error) => expect(error).toMatchObject({ name: 'AbortError' })
|
||||
)
|
||||
controller.abort()
|
||||
canceled.push(request)
|
||||
}
|
||||
await Promise.all(canceled)
|
||||
|
||||
expect(scheduler.snapshot().queued).toBe(0)
|
||||
expect(scheduler.snapshot().candidateCount).toBeLessThanOrEqual(64)
|
||||
running.release()
|
||||
expect(scheduler.snapshot().candidateCount).toBe(0)
|
||||
})
|
||||
|
||||
it('does not decrement for an already-aborted signal with a free slot', async () => {
|
||||
const scheduler = schedulerWithOneSlot()
|
||||
const controller = new AbortController()
|
||||
@@ -293,7 +315,12 @@ describe('GitAdmissionScheduler', () => {
|
||||
await expect(
|
||||
scheduler.acquire({ ...local(), signal: controller.signal })
|
||||
).rejects.toMatchObject({ name: 'AbortError' })
|
||||
expect(scheduler.snapshot()).toEqual({ queued: 0, queuedWaiters: [], budgets: {} })
|
||||
expect(scheduler.snapshot()).toEqual({
|
||||
queued: 0,
|
||||
queuedWaiters: [],
|
||||
candidateCount: 0,
|
||||
budgets: {}
|
||||
})
|
||||
})
|
||||
|
||||
it('returns a grant selected in the same tick when abort wins delivery', async () => {
|
||||
|
||||
@@ -92,11 +92,13 @@ export class GitAdmissionScheduler {
|
||||
queued: number
|
||||
queuedWaiters: { id: number; args: readonly string[]; tier: AdmissionWaiter['tier'] }[]
|
||||
budgets: Record<string, { baseUsed: number; headroomUsed: number }>
|
||||
candidateCount: number
|
||||
} {
|
||||
const queuedWaiters = this.waiters.snapshot()
|
||||
return {
|
||||
queued: this.waiters.count,
|
||||
queuedWaiters: queuedWaiters.map(({ id, args, tier }) => ({ id, args, tier })),
|
||||
candidateCount: this.waiters.candidateCountForTests,
|
||||
budgets: Object.fromEntries(
|
||||
[...this.budgets].map(([key, budget]) => [
|
||||
key,
|
||||
|
||||
Reference in New Issue
Block a user