Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | 140x 140x 140x 140x 140x 146x 146x 98x 98x 48x 46x 46x 46x 1x 1x 1x 46x 46x 143x 143x 143x 143x 143x 143x 45x 45x 98x 6x 35x 35x 35x 35x 35x 70x 2x 35x 70x 35x 35x 123x 120x 120x 120x 120x 15x 15x 15x 15x 15x 15x 15x 2x 120x 120x 120x 120x 120x 11x 109x | import { AsyncLocalStorage } from "node:async_hooks";
import { ExamplesError } from "./examples-error";
import type { SourcePriority } from "./github-api-budget";
interface Waiter {
resolve: (release: () => void) => void;
reject: (error: Error) => void;
signal?: AbortSignal;
onAbort?: () => void;
}
class BoundedSemaphore {
private active = 0;
private readonly queue: Waiter[] = [];
constructor(
private readonly capacity: number,
private readonly maxQueue: number,
private readonly capacityError: () => Error,
) {}
async acquire(signal?: AbortSignal): Promise<() => void> {
Iif (signal?.aborted) throw signal.reason ?? new Error("Request aborted");
if (this.active < this.capacity) {
this.active++;
return this.releaseOnce();
}
if (this.queue.length >= this.maxQueue) throw this.capacityError();
return new Promise<() => void>((resolve, reject) => {
const waiter: Waiter = { resolve, reject, signal };
waiter.onAbort = () => {
const index = this.queue.indexOf(waiter);
Eif (index >= 0) this.queue.splice(index, 1);
reject(signal?.reason ?? new Error("Request aborted"));
};
signal?.addEventListener("abort", waiter.onAbort, { once: true });
this.queue.push(waiter);
});
}
private releaseOnce(): () => void {
let released = false;
return () => {
Iif (released) return;
released = true;
const waiter = this.queue.shift();
if (waiter) {
waiter.signal?.removeEventListener("abort", waiter.onAbort!);
waiter.resolve(this.releaseOnce());
} else {
this.active--;
}
};
}
getStats(): { active: number; queued: number } {
return { active: this.active, queued: this.queue.length };
}
}
export interface ExamplesLoadControllerOptions {
maxConcurrentLoads: number;
maxLoadQueue: number;
maxOutboundFetches: number;
globalLoadStartsPerMinute: number;
now?: () => number;
}
/** Outbound fetches kept for loads of the operator's default Course. */
function defaultOutboundReserve(maxOutboundFetches: number): number {
return maxOutboundFetches > 1 ? Math.max(1, Math.floor(maxOutboundFetches / 4)) : 0;
}
/**
* Bounds remote examples loading. Loads of the operator's default Course and
* browser overrides, which any client can trigger with arbitrary repositories
* and refs, use separate load slots, queues and load-start budgets of the same
* size, so overrides can never deny the default a load. Outbound fetches share
* one cap, but override loads may hold only part of it; the rest is reserved
* for default loads.
*/
export class ExamplesLoadController {
private readonly loads: Record<SourcePriority, BoundedSemaphore>;
private readonly outbound: BoundedSemaphore;
private readonly overrideOutbound: BoundedSemaphore;
private readonly starts: Record<SourcePriority, number[]> = { default: [], override: [] };
private readonly loadPriority = new AsyncLocalStorage<SourcePriority>();
private readonly now: () => number;
constructor(private readonly options: ExamplesLoadControllerOptions) {
this.now = options.now ?? Date.now;
const loadSlots = () => new BoundedSemaphore(
options.maxConcurrentLoads,
options.maxLoadQueue,
() => new ExamplesError("LOAD_CAPACITY_EXCEEDED", "External examples load capacity is exhausted"),
);
this.loads = { default: loadSlots(), override: loadSlots() };
const fetchSlots = (capacity: number) => new BoundedSemaphore(
capacity,
Number.MAX_SAFE_INTEGER,
() => new ExamplesError("LOAD_CAPACITY_EXCEEDED", "External examples fetch capacity is exhausted"),
);
this.outbound = fetchSlots(options.maxOutboundFetches);
this.overrideOutbound = fetchSlots(options.maxOutboundFetches - defaultOutboundReserve(options.maxOutboundFetches));
}
/** Runs one load; its outbound fetches inherit the priority. */
async runLoad<T>(operation: () => Promise<T>, signal?: AbortSignal, priority: SourcePriority = "override"): Promise<T> {
const release = await this.loads[priority].acquire(signal);
try {
this.recordLoadStart(priority);
return await this.loadPriority.run(priority, operation);
} finally {
release();
}
}
async runOutbound<T>(operation: () => Promise<T>, signal?: AbortSignal): Promise<T> {
const releaseShare = this.loadPriority.getStore() === "override"
? await this.overrideOutbound.acquire(signal)
: undefined;
try {
const release = await this.outbound.acquire(signal);
try {
return await operation();
} finally {
release();
}
} finally {
releaseShare?.();
}
}
getStats() {
return {
loads: this.loads.override.getStats(),
defaultLoads: this.loads.default.getStats(),
outbound: this.outbound.getStats(),
};
}
private recordLoadStart(priority: SourcePriority): void {
const starts = this.starts[priority];
const now = this.now();
const cutoff = now - 60_000;
while (starts[0] !== undefined && starts[0] <= cutoff) starts.shift();
if (starts.length >= this.options.globalLoadStartsPerMinute) {
throw new ExamplesError("RATE_LIMITED", "External examples load rate exceeded", 60);
}
starts.push(now);
}
}
|