fix(explore): shrink a later cluster into the remainder instead of dropping it (CG-36)
A file's ranked clusters were all-or-nothing past the first one: the top-ranked cluster was taken (shrunk to fit when it had to be) and every cluster below it was rendered whole, then either fit the remainder or was dropped entirely. On a file whose top-ranked cluster is TRIVIAL that discards the answer — django's `db/models/sql/query.py` kept a 22-line glue cluster and dropped the 624-line `Query` body, spending 1,923 of a 7,947 reservation; okhttp's `RealInterceptorChain.kt` did the same behind its import header. The response stayed full, which is why this was invisible: the unspent reservation carried forward exactly as designed and a file scoring a fifth as much took the bytes. Two sites, the same rule — hold the remainder while it is still worth a section (CG-26's between-FILES lesson, applied between CLUSTERS): - selection now shrinks a later cluster into what is left of the file's budget, by the same whole-member rule the first cluster already used; - the ceiling trim re-renders the weakest cluster into the room that remains before dropping it. On excalidraw's `typeChecks.ts` the section-cost estimate missed by 13 chars and a 1,512-char cluster — the file's highest-SCORING one — was thrown away to pay for it. Cluster RANKING is untouched: measured, both real cases lost on `maxImportance`, not on the density tiebreak the issue suspected, and density-first is what keeps Alamofire's `Session.swift` from burying its methods under the property list. Suite (6 repos, clean-rebuilt indexes): all 8 starvation flags cleared, +1,012 source chars net. django's `sql/query.py` 1,923 -> 10,082 of 7,947, okhttp's `RealInterceptorChain.kt` 1,474 -> 6,038 of 6,058, gin's `routergroup.go` 3,273 -> 5,632. okhttp trades its rank-6 file (score 21) for +7,196 chars in the two files that answer the question. Ships two fixtures pulling in opposite directions (`starved-cluster-ts` and `dense-header-ts`), a `spendShareAtLeast` gate in probe-allocation, and probe-file-spend.mjs — a standing per-file reservation-vs-delivered sweep.
This commit is contained in:
Colby McHenry
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"name": "dense-header-fixture",
|
||||
"private": true,
|
||||
"version": "0.0.0",
|
||||
"type": "module"
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
import type { URLSessionTask } from './types';
|
||||
|
||||
export class RequestQueue {
|
||||
private readonly waiting: URLSessionTask[] = [];
|
||||
private running = 0;
|
||||
|
||||
enqueue(task: URLSessionTask, limit: number): void {
|
||||
if (this.running < limit) {
|
||||
this.running += 1;
|
||||
return;
|
||||
}
|
||||
this.waiting.push(task);
|
||||
}
|
||||
|
||||
release(): URLSessionTask | undefined {
|
||||
this.running = Math.max(0, this.running - 1);
|
||||
return this.waiting.shift();
|
||||
}
|
||||
|
||||
get depth(): number {
|
||||
return this.waiting.length;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
import type { CachePolicy, URLRequest } from './types';
|
||||
|
||||
export function buildURLRequest(options: {
|
||||
url: string;
|
||||
method: string;
|
||||
body?: Uint8Array;
|
||||
headers: Record<string, string>;
|
||||
timeout: number;
|
||||
cachePolicy: CachePolicy;
|
||||
}): URLRequest {
|
||||
const headers = { ...options.headers };
|
||||
if (options.body && !headers['content-length']) {
|
||||
headers['content-length'] = String(options.body.length);
|
||||
}
|
||||
return {
|
||||
url: normalize(options.url),
|
||||
method: options.method.toUpperCase(),
|
||||
headers,
|
||||
body: options.body,
|
||||
timeout: options.timeout,
|
||||
cachePolicy: options.cachePolicy,
|
||||
};
|
||||
}
|
||||
|
||||
function normalize(url: string): string {
|
||||
return url.endsWith('/') && url.split('/').length > 4 ? url.slice(0, -1) : url;
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
import type { RequestDelegate, TaskResponse, URLRequest, URLSessionTask } from './types';
|
||||
|
||||
export function makeTask(options: {
|
||||
identifier: number;
|
||||
request: URLRequest;
|
||||
delegate: RequestDelegate;
|
||||
allowsCellularAccess: boolean;
|
||||
waitsForConnectivity: boolean;
|
||||
resourceTimeout: number;
|
||||
}): URLSessionTask {
|
||||
const handlers: Array<(response: TaskResponse) => void> = [];
|
||||
return {
|
||||
identifier: options.identifier,
|
||||
request: options.request,
|
||||
state: 'initialized',
|
||||
cancel() { this.state = 'cancelled'; },
|
||||
onComplete(handler) { handlers.push(handler); },
|
||||
};
|
||||
}
|
||||
|
||||
export function resumeTask(task: URLSessionTask): void {
|
||||
task.state = 'resumed';
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
export type CachePolicy = 'useProtocolCachePolicy' | 'reloadIgnoringLocalCacheData' | 'returnCacheDataElseLoad';
|
||||
export type RequestState = 'initialized' | 'resumed' | 'suspended' | 'cancelled' | 'finished';
|
||||
|
||||
export interface URLRequest {
|
||||
url: string;
|
||||
method: string;
|
||||
headers: Record<string, string>;
|
||||
body?: Uint8Array;
|
||||
timeout: number;
|
||||
cachePolicy: CachePolicy;
|
||||
}
|
||||
|
||||
export interface TaskResponse {
|
||||
status: number;
|
||||
headers: Record<string, string>;
|
||||
body: Uint8Array;
|
||||
}
|
||||
|
||||
export interface URLSessionTask {
|
||||
identifier: number;
|
||||
request: URLRequest;
|
||||
state: RequestState;
|
||||
cancel(): void;
|
||||
onComplete(handler: (response: TaskResponse) => void): void;
|
||||
}
|
||||
|
||||
export interface Adapter { adapt(request: URLRequest): URLRequest; }
|
||||
export interface Serializer { serialize(value: unknown): Uint8Array; }
|
||||
export interface Validator { validate(response: TaskResponse): { ok: boolean; reason?: string }; }
|
||||
export interface Retrier { shouldRetry(response: TaskResponse, verdict: { ok: boolean }): boolean; }
|
||||
export interface RedirectHandler { resolve(location: string, original: URLRequest): { url: string; method: string; body?: Uint8Array } | null; }
|
||||
export interface TrustEvaluator { evaluate(host: string): boolean; }
|
||||
export interface Credential { apply(request: URLRequest): URLRequest; }
|
||||
export interface Interceptor { name: string; adapt(request: URLRequest, session: unknown): Promise<URLRequest>; }
|
||||
export interface RequestDelegate { willSend(request: URLRequest): void; }
|
||||
export interface EventMonitor {
|
||||
didAdaptRequest(request: URLRequest, interceptor: string): void;
|
||||
didCreateTask(task: URLSessionTask, request: URLRequest): void;
|
||||
didResumeTask(task: URLSessionTask): void;
|
||||
didRetryTask(task: URLSessionTask, previousIdentifier: number): void;
|
||||
didCompleteTask(task: URLSessionTask, response: TaskResponse): void;
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
export { Session } from './net/session';
|
||||
export { RequestQueue } from './core/queue';
|
||||
export { buildURLRequest } from './core/request-builder';
|
||||
@@ -0,0 +1,285 @@
|
||||
import type {
|
||||
Adapter,
|
||||
CachePolicy,
|
||||
Credential,
|
||||
EventMonitor,
|
||||
Interceptor,
|
||||
RedirectHandler,
|
||||
RequestDelegate,
|
||||
RequestState,
|
||||
Retrier,
|
||||
Serializer,
|
||||
TrustEvaluator,
|
||||
URLRequest,
|
||||
URLSessionTask,
|
||||
Validator,
|
||||
} from '../core/types';
|
||||
import { buildURLRequest } from '../core/request-builder';
|
||||
import { makeTask, resumeTask } from '../core/task-factory';
|
||||
import { RequestQueue } from '../core/queue';
|
||||
|
||||
/**
|
||||
* The shape density-first ranking exists for: a class whose top-of-file header
|
||||
* is a long, tightly-packed property list — dozens of adjacent declarations,
|
||||
* each individually trivial — while the methods a flow question actually asks
|
||||
* about live hundreds of lines below it.
|
||||
*
|
||||
* Ranked by density alone the header wins the file's whole budget and the
|
||||
* methods are buried. The ranking puts importance first for exactly this
|
||||
* reason, and density only breaks ties inside one importance tier.
|
||||
*/
|
||||
export class Session {
|
||||
readonly identifier: string;
|
||||
readonly adapter: Adapter;
|
||||
readonly serializer: Serializer;
|
||||
readonly validator: Validator;
|
||||
readonly retrier: Retrier;
|
||||
readonly redirectHandler: RedirectHandler;
|
||||
readonly trustEvaluator: TrustEvaluator;
|
||||
readonly eventMonitor: EventMonitor;
|
||||
readonly cachePolicy: CachePolicy;
|
||||
readonly credential: Credential | null;
|
||||
readonly interceptors: Interceptor[];
|
||||
readonly delegate: RequestDelegate;
|
||||
readonly queue: RequestQueue;
|
||||
readonly startRequestsImmediately: boolean;
|
||||
readonly maximumConnectionsPerHost: number;
|
||||
readonly timeoutIntervalForRequest: number;
|
||||
readonly timeoutIntervalForResource: number;
|
||||
readonly allowsCellularAccess: boolean;
|
||||
readonly waitsForConnectivity: boolean;
|
||||
readonly httpShouldUsePipelining: boolean;
|
||||
readonly httpShouldSetCookies: boolean;
|
||||
readonly httpMaximumConnectionsPerHost: number;
|
||||
readonly sessionConfigurationName: string;
|
||||
readonly requestState: RequestState;
|
||||
readonly defaultHeaders: Record<string, string>;
|
||||
readonly userAgent: string;
|
||||
readonly acceptEncoding: string;
|
||||
readonly acceptLanguage: string;
|
||||
private taskCounter = 0;
|
||||
private active = new Map<number, URLSessionTask>();
|
||||
|
||||
constructor(options: Partial<Session> & { identifier: string }) {
|
||||
this.identifier = options.identifier;
|
||||
this.adapter = options.adapter!;
|
||||
this.serializer = options.serializer!;
|
||||
this.validator = options.validator!;
|
||||
this.retrier = options.retrier!;
|
||||
this.redirectHandler = options.redirectHandler!;
|
||||
this.trustEvaluator = options.trustEvaluator!;
|
||||
this.eventMonitor = options.eventMonitor!;
|
||||
this.cachePolicy = options.cachePolicy ?? 'useProtocolCachePolicy';
|
||||
this.credential = options.credential ?? null;
|
||||
this.interceptors = options.interceptors ?? [];
|
||||
this.delegate = options.delegate!;
|
||||
this.queue = options.queue ?? new RequestQueue();
|
||||
this.startRequestsImmediately = options.startRequestsImmediately ?? true;
|
||||
this.maximumConnectionsPerHost = options.maximumConnectionsPerHost ?? 6;
|
||||
this.timeoutIntervalForRequest = options.timeoutIntervalForRequest ?? 60;
|
||||
this.timeoutIntervalForResource = options.timeoutIntervalForResource ?? 604800;
|
||||
this.allowsCellularAccess = options.allowsCellularAccess ?? true;
|
||||
this.waitsForConnectivity = options.waitsForConnectivity ?? false;
|
||||
this.httpShouldUsePipelining = options.httpShouldUsePipelining ?? false;
|
||||
this.httpShouldSetCookies = options.httpShouldSetCookies ?? true;
|
||||
this.httpMaximumConnectionsPerHost = options.httpMaximumConnectionsPerHost ?? 6;
|
||||
this.sessionConfigurationName = options.sessionConfigurationName ?? 'default';
|
||||
this.requestState = options.requestState ?? 'initialized';
|
||||
this.defaultHeaders = options.defaultHeaders ?? {};
|
||||
this.userAgent = options.userAgent ?? 'session/1.0';
|
||||
this.acceptEncoding = options.acceptEncoding ?? 'br;q=1.0, gzip;q=0.9';
|
||||
this.acceptLanguage = options.acceptLanguage ?? 'en;q=1.0';
|
||||
}
|
||||
|
||||
// -- configuration accessors ----------------------------------------------
|
||||
// Individually trivial, adjacent, and dense. On the density tiebreak alone
|
||||
// this block outranks anything with a body worth reading.
|
||||
|
||||
get isBackground(): boolean {
|
||||
return this.sessionConfigurationName === 'background';
|
||||
}
|
||||
|
||||
get connectionLimit(): number {
|
||||
return Math.min(this.maximumConnectionsPerHost, this.httpMaximumConnectionsPerHost);
|
||||
}
|
||||
|
||||
get headerDefaults(): Record<string, string> {
|
||||
return { ...this.defaultHeaders, 'user-agent': this.userAgent };
|
||||
}
|
||||
|
||||
get acceptHeaders(): Record<string, string> {
|
||||
return { 'accept-encoding': this.acceptEncoding, 'accept-language': this.acceptLanguage };
|
||||
}
|
||||
|
||||
get activeCount(): number {
|
||||
return this.active.size;
|
||||
}
|
||||
|
||||
get isIdle(): boolean {
|
||||
return this.active.size === 0;
|
||||
}
|
||||
|
||||
get nextIdentifier(): number {
|
||||
return this.taskCounter + 1;
|
||||
}
|
||||
|
||||
get description(): string {
|
||||
return `Session(${this.identifier}, ${this.sessionConfigurationName})`;
|
||||
}
|
||||
|
||||
cancelAll(): void {
|
||||
for (const task of this.active.values()) task.cancel();
|
||||
this.active.clear();
|
||||
}
|
||||
|
||||
taskFor(identifier: number): URLSessionTask | undefined {
|
||||
return this.active.get(identifier);
|
||||
}
|
||||
|
||||
headers(): Record<string, string> {
|
||||
return { ...this.headerDefaults, ...this.acceptHeaders };
|
||||
}
|
||||
|
||||
withUserAgent(userAgent: string): Session {
|
||||
return new Session({ ...this, identifier: this.identifier, userAgent });
|
||||
}
|
||||
|
||||
withTimeout(seconds: number): Session {
|
||||
return new Session({ ...this, identifier: this.identifier, timeoutIntervalForRequest: seconds });
|
||||
}
|
||||
|
||||
withInterceptor(interceptor: Interceptor): Session {
|
||||
return new Session({
|
||||
...this,
|
||||
identifier: this.identifier,
|
||||
interceptors: [...this.interceptors, interceptor],
|
||||
});
|
||||
}
|
||||
|
||||
withCredential(credential: Credential): Session {
|
||||
return new Session({ ...this, identifier: this.identifier, credential });
|
||||
}
|
||||
|
||||
withCachePolicy(cachePolicy: CachePolicy): Session {
|
||||
return new Session({ ...this, identifier: this.identifier, cachePolicy });
|
||||
}
|
||||
|
||||
withQueue(queue: RequestQueue): Session {
|
||||
return new Session({ ...this, identifier: this.identifier, queue });
|
||||
}
|
||||
|
||||
withAdapter(adapter: Adapter): Session {
|
||||
return new Session({ ...this, identifier: this.identifier, adapter });
|
||||
}
|
||||
|
||||
withValidator(validator: Validator): Session {
|
||||
return new Session({ ...this, identifier: this.identifier, validator });
|
||||
}
|
||||
|
||||
withRetrier(retrier: Retrier): Session {
|
||||
return new Session({ ...this, identifier: this.identifier, retrier });
|
||||
}
|
||||
|
||||
withMonitor(eventMonitor: EventMonitor): Session {
|
||||
return new Session({ ...this, identifier: this.identifier, eventMonitor });
|
||||
}
|
||||
|
||||
// -- the flow ---------------------------------------------------------------
|
||||
//
|
||||
// The methods below are what a "how does a request get built and sent" question
|
||||
// is about, and they sit hundreds of lines under the header block.
|
||||
|
||||
/**
|
||||
* Turn a convenience call into a URLRequest, hand it to the adapter chain and
|
||||
* start the resulting task. The entry point of the whole flow.
|
||||
*/
|
||||
async perform(url: string, method: string, body?: Uint8Array): Promise<URLSessionTask> {
|
||||
const initial = buildURLRequest({
|
||||
url,
|
||||
method,
|
||||
body,
|
||||
headers: this.headers(),
|
||||
timeout: this.timeoutIntervalForRequest,
|
||||
cachePolicy: this.cachePolicy,
|
||||
});
|
||||
const adapted = await this.adapt(initial);
|
||||
return this.didCreateURLRequest(adapted);
|
||||
}
|
||||
|
||||
/**
|
||||
* Every interceptor gets a chance to rewrite the request before it becomes a
|
||||
* task. Runs in registration order, and a thrown error aborts the whole call.
|
||||
*/
|
||||
private async adapt(request: URLRequest): Promise<URLRequest> {
|
||||
let current = request;
|
||||
for (const interceptor of this.interceptors) {
|
||||
current = await interceptor.adapt(current, this);
|
||||
this.eventMonitor.didAdaptRequest(current, interceptor.name);
|
||||
}
|
||||
if (this.credential) current = this.credential.apply(current);
|
||||
return current;
|
||||
}
|
||||
|
||||
/**
|
||||
* The adapted request is final: build the task around it, register it and —
|
||||
* unless the session was told to wait — resume it immediately.
|
||||
*/
|
||||
didCreateURLRequest(request: URLRequest): URLSessionTask {
|
||||
this.taskCounter += 1;
|
||||
const identifier = this.taskCounter;
|
||||
const created = this.task(request, identifier);
|
||||
this.active.set(identifier, created);
|
||||
this.eventMonitor.didCreateTask(created, request);
|
||||
if (this.startRequestsImmediately) this.resume(created);
|
||||
return created;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the URLSessionTask for a request. Split out from
|
||||
* `didCreateURLRequest` because retries rebuild the task without going back
|
||||
* through the adapter chain.
|
||||
*/
|
||||
task(request: URLRequest, identifier: number): URLSessionTask {
|
||||
const created = makeTask({
|
||||
identifier,
|
||||
request,
|
||||
delegate: this.delegate,
|
||||
allowsCellularAccess: this.allowsCellularAccess,
|
||||
waitsForConnectivity: this.waitsForConnectivity,
|
||||
resourceTimeout: this.timeoutIntervalForResource,
|
||||
});
|
||||
created.onComplete((response) => {
|
||||
this.active.delete(identifier);
|
||||
const verdict = this.validator.validate(response);
|
||||
if (!verdict.ok && this.retrier.shouldRetry(response, verdict)) {
|
||||
this.retry(request, identifier);
|
||||
return;
|
||||
}
|
||||
this.eventMonitor.didCompleteTask(created, response);
|
||||
});
|
||||
return created;
|
||||
}
|
||||
|
||||
/** Put a built task on the queue and start it. */
|
||||
resume(task: URLSessionTask): void {
|
||||
this.queue.enqueue(task, this.connectionLimit);
|
||||
resumeTask(task);
|
||||
this.eventMonitor.didResumeTask(task);
|
||||
}
|
||||
|
||||
/** Rebuild and restart a task the retrier asked for. */
|
||||
private retry(request: URLRequest, previousIdentifier: number): void {
|
||||
this.taskCounter += 1;
|
||||
const retried = this.task(request, this.taskCounter);
|
||||
this.active.set(this.taskCounter, retried);
|
||||
this.eventMonitor.didRetryTask(retried, previousIdentifier);
|
||||
this.resume(retried);
|
||||
}
|
||||
|
||||
/** Follow a redirect by adapting and re-performing the new location. */
|
||||
async follow(response: { location: string }, original: URLRequest): Promise<URLSessionTask> {
|
||||
const target = this.redirectHandler.resolve(response.location, original);
|
||||
if (!target) throw new Error(`redirect to ${response.location} refused`);
|
||||
return this.perform(target.url, target.method, target.body);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"name": "starved-cluster-fixture",
|
||||
"private": true,
|
||||
"version": "0.0.0",
|
||||
"type": "module"
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
import { RequestChain, describeChain } from '../pipeline/chain';
|
||||
import type { PipelineRequest, PipelineResponse } from '../pipeline/types';
|
||||
import { openSocket } from '../transport/socket';
|
||||
|
||||
/**
|
||||
* The entry point a caller reaches for. Everything the chain does happens
|
||||
* underneath this call, which is why a flow question names it.
|
||||
*/
|
||||
export async function sendRequest(request: PipelineRequest): Promise<PipelineResponse> {
|
||||
const socket = openSocket(request.host, request.port);
|
||||
const chain = new RequestChain(request, socket);
|
||||
trace(describeChain(chain));
|
||||
return chain.proceed(request);
|
||||
}
|
||||
|
||||
export function trace(line: string): void {
|
||||
if (process.env.PIPELINE_TRACE) process.stderr.write(`${line}\n`);
|
||||
}
|
||||
|
||||
export async function sendAll(requests: PipelineRequest[]): Promise<PipelineResponse[]> {
|
||||
const out: PipelineResponse[] = [];
|
||||
for (const request of requests) out.push(await sendRequest(request));
|
||||
return out;
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
export interface ClientConfig {
|
||||
host: string;
|
||||
port: number;
|
||||
retries: number;
|
||||
userAgent: string;
|
||||
}
|
||||
|
||||
export function defaultConfig(): ClientConfig {
|
||||
return { host: 'localhost', port: 8080, retries: 3, userAgent: 'pipeline/1.0' };
|
||||
}
|
||||
|
||||
export function withHost(config: ClientConfig, host: string): ClientConfig {
|
||||
return { ...config, host };
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
export { sendRequest, sendAll } from './app/client';
|
||||
export { RequestChain, describeChain } from './pipeline/chain';
|
||||
export { openSocket } from './transport/socket';
|
||||
export { defaultConfig } from './app/config';
|
||||
@@ -0,0 +1,318 @@
|
||||
import type { PipelineRequest, PipelineResponse, Interceptor, Socket } from './types';
|
||||
import { encodeFrame, decodeFrame } from './framing';
|
||||
import { defaultInterceptors } from './interceptors';
|
||||
|
||||
/**
|
||||
* A one-line summary of a chain, used only by the tracing hook in the caller.
|
||||
* It is TRIVIAL — it answers nothing about how a request travels — but it sits
|
||||
* next to the entry point in the call graph, so its cluster carries the file's
|
||||
* highest per-symbol importance.
|
||||
*/
|
||||
export function describeChain(chain: RequestChain): string {
|
||||
return `chain(${chain.index}/${chain.size}) -> ${chain.hostLabel}`;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
//
|
||||
// Everything below is the part a "how does a request reach the socket" question
|
||||
// is actually asking about. It is separated from the helper above by more than
|
||||
// the cluster gap threshold, so it forms its own cluster — a large one, whose
|
||||
// symbols are reached transitively rather than named.
|
||||
//
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export class RequestChain {
|
||||
readonly index: number;
|
||||
readonly size: number;
|
||||
readonly hostLabel: string;
|
||||
private readonly interceptors: Interceptor[];
|
||||
private readonly socket: Socket;
|
||||
private readonly request: PipelineRequest;
|
||||
private connectTimeoutMs = 10_000;
|
||||
private readTimeoutMs = 10_000;
|
||||
private writeTimeoutMs = 10_000;
|
||||
private calls = 0;
|
||||
|
||||
constructor(request: PipelineRequest, socket: Socket, index = 0, interceptors?: Interceptor[]) {
|
||||
this.request = request;
|
||||
this.socket = socket;
|
||||
this.index = index;
|
||||
this.interceptors = interceptors ?? defaultInterceptors();
|
||||
this.size = this.interceptors.length;
|
||||
this.hostLabel = `${request.host}:${request.port}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the request through the remaining interceptors and, once they are
|
||||
* exhausted, hand it to the transport. This is the method the flow question
|
||||
* is about: every hop between the caller and the socket passes through here.
|
||||
*/
|
||||
async proceed(request: PipelineRequest): Promise<PipelineResponse> {
|
||||
if (this.index >= this.size) {
|
||||
return this.writeAndRead(request);
|
||||
}
|
||||
this.calls += 1;
|
||||
if (this.calls > 1) {
|
||||
throw new Error(`chain link ${this.index} called ${this.calls} times`);
|
||||
}
|
||||
const next = this.advance(request);
|
||||
const interceptor = this.interceptors[this.index]!;
|
||||
const response = await interceptor.intercept(next);
|
||||
if (!response) {
|
||||
throw new Error(`interceptor ${interceptor.name} returned no response`);
|
||||
}
|
||||
if (this.index + 1 < this.size && next.callCount() === 0) {
|
||||
throw new Error(`interceptor ${interceptor.name} must call proceed()`);
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
/**
|
||||
* The next link in the chain: the same chain with the cursor moved on and the
|
||||
* timeouts carried over. Cloning here is what keeps each interceptor from
|
||||
* mutating the chain the one before it is still holding.
|
||||
*/
|
||||
advance(request: PipelineRequest): RequestChain {
|
||||
const next = new RequestChain(request, this.socket, this.index + 1, this.interceptors);
|
||||
next.connectTimeoutMs = this.connectTimeoutMs;
|
||||
next.readTimeoutMs = this.readTimeoutMs;
|
||||
next.writeTimeoutMs = this.writeTimeoutMs;
|
||||
return next;
|
||||
}
|
||||
|
||||
callCount(): number {
|
||||
return this.calls;
|
||||
}
|
||||
|
||||
/**
|
||||
* The end of the chain: frame the request, put the bytes on the socket, wait
|
||||
* for the reply and decode it. Past this point there is no more pipeline —
|
||||
* this is the transport hop the question is looking for.
|
||||
*/
|
||||
private async writeAndRead(request: PipelineRequest): Promise<PipelineResponse> {
|
||||
const frame = encodeFrame(request);
|
||||
await this.socket.connect(this.connectTimeoutMs);
|
||||
await this.socket.write(frame, this.writeTimeoutMs);
|
||||
const raw = await this.socket.read(this.readTimeoutMs);
|
||||
const decoded = decodeFrame(raw);
|
||||
return {
|
||||
status: decoded.status,
|
||||
headers: decoded.headers,
|
||||
body: decoded.body,
|
||||
request,
|
||||
};
|
||||
}
|
||||
|
||||
withConnectTimeout(ms: number): RequestChain {
|
||||
const next = this.advance(this.request);
|
||||
next.connectTimeoutMs = checkDuration('connectTimeout', ms);
|
||||
return next;
|
||||
}
|
||||
|
||||
withReadTimeout(ms: number): RequestChain {
|
||||
const next = this.advance(this.request);
|
||||
next.readTimeoutMs = checkDuration('readTimeout', ms);
|
||||
return next;
|
||||
}
|
||||
|
||||
withWriteTimeout(ms: number): RequestChain {
|
||||
const next = this.advance(this.request);
|
||||
next.writeTimeoutMs = checkDuration('writeTimeout', ms);
|
||||
return next;
|
||||
}
|
||||
|
||||
connectTimeout(): number {
|
||||
return this.connectTimeoutMs;
|
||||
}
|
||||
|
||||
readTimeout(): number {
|
||||
return this.readTimeoutMs;
|
||||
}
|
||||
|
||||
writeTimeout(): number {
|
||||
return this.writeTimeoutMs;
|
||||
}
|
||||
|
||||
/**
|
||||
* Retry policy for the transport hop. Sits inside the same cluster as the
|
||||
* proceed/advance pair, so it is part of what a shrink has to choose between.
|
||||
*/
|
||||
async retryWrite(request: PipelineRequest, attempts: number): Promise<PipelineResponse> {
|
||||
let lastError: unknown;
|
||||
for (let attempt = 0; attempt < attempts; attempt += 1) {
|
||||
try {
|
||||
return await this.writeAndRead(request);
|
||||
} catch (error) {
|
||||
lastError = error;
|
||||
await backoff(attempt);
|
||||
}
|
||||
}
|
||||
throw lastError;
|
||||
}
|
||||
|
||||
/** Whether the chain may still be resumed after a transport failure. */
|
||||
canRetry(error: unknown): boolean {
|
||||
if (this.index >= this.size) return false;
|
||||
if (!(error instanceof Error)) return false;
|
||||
return error.message.includes('timeout') || error.message.includes('reset');
|
||||
}
|
||||
|
||||
/** The interceptor names, in the order the request will visit them. */
|
||||
route(): string[] {
|
||||
return this.interceptors.slice(this.index).map((i) => i.name);
|
||||
}
|
||||
|
||||
/** A copy of the chain rewound to the first interceptor. */
|
||||
rewind(): RequestChain {
|
||||
return new RequestChain(this.request, this.socket, 0, this.interceptors);
|
||||
}
|
||||
|
||||
/** Drop one interceptor by name and return the shortened chain. */
|
||||
without(name: string): RequestChain {
|
||||
const kept = this.interceptors.filter((i) => i.name !== name);
|
||||
return new RequestChain(this.request, this.socket, this.index, kept);
|
||||
}
|
||||
|
||||
/** Append an interceptor to the end of the chain. */
|
||||
with(interceptor: Interceptor): RequestChain {
|
||||
return new RequestChain(
|
||||
this.request,
|
||||
this.socket,
|
||||
this.index,
|
||||
[...this.interceptors, interceptor],
|
||||
);
|
||||
}
|
||||
|
||||
/** Close the transport this chain was built around. */
|
||||
async close(): Promise<void> {
|
||||
await this.socket.close();
|
||||
}
|
||||
|
||||
/** Headers the transport hop will actually put on the wire. */
|
||||
effectiveHeaders(): Record<string, string> {
|
||||
const headers: Record<string, string> = { ...this.request.headers };
|
||||
headers['host'] = this.hostLabel;
|
||||
headers['x-chain-index'] = String(this.index);
|
||||
headers['x-chain-size'] = String(this.size);
|
||||
if (this.request.body) headers['content-length'] = String(this.request.body.length);
|
||||
return headers;
|
||||
}
|
||||
|
||||
/** The request as the next link will see it, with the chain's headers merged. */
|
||||
prepared(): PipelineRequest {
|
||||
return { ...this.request, headers: this.effectiveHeaders() };
|
||||
}
|
||||
|
||||
/**
|
||||
* Send the prepared request through the rest of the chain. The convenience
|
||||
* wrapper most callers use instead of building the request themselves.
|
||||
*/
|
||||
async send(): Promise<PipelineResponse> {
|
||||
return this.proceed(this.prepared());
|
||||
}
|
||||
|
||||
/** Whether the chain has any interceptor left before the transport hop. */
|
||||
hasNext(): boolean {
|
||||
return this.index < this.size;
|
||||
}
|
||||
|
||||
/** The interceptor the next `proceed` will run, if there is one. */
|
||||
peek(): Interceptor | undefined {
|
||||
return this.interceptors[this.index];
|
||||
}
|
||||
|
||||
/** Total configured wait for one attempt, across all three timeouts. */
|
||||
totalTimeout(): number {
|
||||
return this.connectTimeoutMs + this.readTimeoutMs + this.writeTimeoutMs;
|
||||
}
|
||||
|
||||
/** Apply one timeout budget to all three phases at once. */
|
||||
withTimeout(ms: number): RequestChain {
|
||||
const next = this.advance(this.request);
|
||||
const checked = checkDuration('timeout', ms);
|
||||
next.connectTimeoutMs = checked;
|
||||
next.readTimeoutMs = checked;
|
||||
next.writeTimeoutMs = checked;
|
||||
return next;
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the chain and translate a transport failure into a response, so a
|
||||
* caller that only cares about the status code never sees an exception.
|
||||
*/
|
||||
async sendOrStatus(status: number): Promise<PipelineResponse> {
|
||||
try {
|
||||
return await this.send();
|
||||
} catch {
|
||||
return {
|
||||
status,
|
||||
headers: this.effectiveHeaders(),
|
||||
body: new Uint8Array(),
|
||||
request: this.request,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/** A short description of where in the chain this link sits. */
|
||||
position(): string {
|
||||
return `${this.index + 1} of ${this.size + 1}`;
|
||||
}
|
||||
|
||||
/** The chain rebuilt around a different transport. */
|
||||
onSocket(socket: Socket): RequestChain {
|
||||
return new RequestChain(this.request, socket, this.index, this.interceptors);
|
||||
}
|
||||
|
||||
/**
|
||||
* Replay the request through the chain from the start, reusing the transport.
|
||||
* Used when an interceptor decides the response it got is not usable and the
|
||||
* whole pipeline has to run again against the same connection.
|
||||
*/
|
||||
async replay(): Promise<PipelineResponse> {
|
||||
const fresh = this.rewind();
|
||||
try {
|
||||
return await fresh.send();
|
||||
} finally {
|
||||
if (!fresh.hasNext()) await fresh.close();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate the chain before it runs: every interceptor named once, timeouts
|
||||
* inside their bounds, and a transport still open at the end of it.
|
||||
*/
|
||||
validate(): string[] {
|
||||
const problems: string[] = [];
|
||||
const seen = new Set<string>();
|
||||
for (const interceptor of this.interceptors) {
|
||||
if (seen.has(interceptor.name)) problems.push(`duplicate interceptor ${interceptor.name}`);
|
||||
seen.add(interceptor.name);
|
||||
}
|
||||
if (this.connectTimeoutMs <= 0) problems.push('connect timeout must be positive');
|
||||
if (this.readTimeoutMs <= 0) problems.push('read timeout must be positive');
|
||||
if (this.writeTimeoutMs <= 0) problems.push('write timeout must be positive');
|
||||
if (this.index > this.size) problems.push('chain cursor is past the end');
|
||||
return problems;
|
||||
}
|
||||
|
||||
/**
|
||||
* The transport hop on its own, with the chain's timeouts but none of its
|
||||
* interceptors — the escape hatch a caller uses to bypass the pipeline.
|
||||
*/
|
||||
async direct(request: PipelineRequest): Promise<PipelineResponse> {
|
||||
const problems = this.validate();
|
||||
if (problems.length > 0) throw new Error(problems.join('; '));
|
||||
return this.writeAndRead(request);
|
||||
}
|
||||
}
|
||||
|
||||
function checkDuration(name: string, ms: number): number {
|
||||
if (!Number.isFinite(ms) || ms < 0) throw new Error(`${name} must be a positive duration`);
|
||||
if (ms > 24 * 60 * 60 * 1000) throw new Error(`${name} is longer than a day`);
|
||||
return Math.round(ms);
|
||||
}
|
||||
|
||||
async function backoff(attempt: number): Promise<void> {
|
||||
const ms = Math.min(1000, 25 * 2 ** attempt);
|
||||
await new Promise((resolve) => setTimeout(resolve, ms));
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
import type { PipelineRequest } from './types';
|
||||
|
||||
export function encodeFrame(request: PipelineRequest): Uint8Array {
|
||||
const head = `${request.method} ${request.path}\n`;
|
||||
const headers = Object.entries(request.headers).map(([k, v]) => `${k}: ${v}`).join('\n');
|
||||
const text = `${head}${headers}\n\n`;
|
||||
const body = request.body ?? new Uint8Array();
|
||||
const out = new Uint8Array(text.length + body.length);
|
||||
out.set(new TextEncoder().encode(text), 0);
|
||||
out.set(body, text.length);
|
||||
return out;
|
||||
}
|
||||
|
||||
export function decodeFrame(raw: Uint8Array): { status: number; headers: Record<string, string>; body: Uint8Array } {
|
||||
const text = new TextDecoder().decode(raw);
|
||||
const split = text.indexOf('\n\n');
|
||||
const head = split < 0 ? text : text.slice(0, split);
|
||||
const lines = head.split('\n');
|
||||
const status = Number.parseInt(lines[0]?.split(' ')[1] ?? '0', 10);
|
||||
const headers: Record<string, string> = {};
|
||||
for (const line of lines.slice(1)) {
|
||||
const at = line.indexOf(': ');
|
||||
if (at > 0) headers[line.slice(0, at)] = line.slice(at + 2);
|
||||
}
|
||||
return { status, headers, body: raw.slice(split < 0 ? raw.length : split + 2) };
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
import type { Interceptor } from './types';
|
||||
|
||||
export function defaultInterceptors(): Interceptor[] {
|
||||
return [retryInterceptor(), headerInterceptor(), logInterceptor()];
|
||||
}
|
||||
|
||||
export function retryInterceptor(): Interceptor {
|
||||
return { name: 'retry', intercept: (chain) => chain.proceed(currentRequest()) };
|
||||
}
|
||||
|
||||
export function headerInterceptor(): Interceptor {
|
||||
return { name: 'headers', intercept: (chain) => chain.proceed(currentRequest()) };
|
||||
}
|
||||
|
||||
export function logInterceptor(): Interceptor {
|
||||
return { name: 'log', intercept: (chain) => chain.proceed(currentRequest()) };
|
||||
}
|
||||
|
||||
function currentRequest() {
|
||||
return { host: 'localhost', port: 80, method: 'GET', path: '/', headers: {} };
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
export interface PipelineRequest {
|
||||
host: string;
|
||||
port: number;
|
||||
method: string;
|
||||
path: string;
|
||||
headers: Record<string, string>;
|
||||
body?: Uint8Array;
|
||||
}
|
||||
|
||||
export interface PipelineResponse {
|
||||
status: number;
|
||||
headers: Record<string, string>;
|
||||
body: Uint8Array;
|
||||
request: PipelineRequest;
|
||||
}
|
||||
|
||||
export interface Interceptor {
|
||||
name: string;
|
||||
intercept(chain: { proceed(request: PipelineRequest): Promise<PipelineResponse> }): Promise<PipelineResponse>;
|
||||
}
|
||||
|
||||
export interface Socket {
|
||||
connect(timeoutMs: number): Promise<void>;
|
||||
write(frame: Uint8Array, timeoutMs: number): Promise<void>;
|
||||
read(timeoutMs: number): Promise<Uint8Array>;
|
||||
close(): Promise<void>;
|
||||
}
|
||||
@@ -0,0 +1,30 @@
|
||||
import type { Socket } from '../pipeline/types';
|
||||
|
||||
/** Open a transport socket for a host/port pair. */
|
||||
export function openSocket(host: string, port: number): Socket {
|
||||
let open = false;
|
||||
const inbox: Uint8Array[] = [];
|
||||
return {
|
||||
async connect(timeoutMs: number) {
|
||||
if (open) return;
|
||||
await settle(timeoutMs);
|
||||
open = true;
|
||||
},
|
||||
async write(frame: Uint8Array, timeoutMs: number) {
|
||||
if (!open) throw new Error(`socket to ${host}:${port} is not connected`);
|
||||
await settle(timeoutMs);
|
||||
inbox.push(frame);
|
||||
},
|
||||
async read(timeoutMs: number) {
|
||||
await settle(timeoutMs);
|
||||
return inbox.shift() ?? new Uint8Array();
|
||||
},
|
||||
async close() {
|
||||
open = false;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function settle(timeoutMs: number): Promise<void> {
|
||||
if (timeoutMs <= 0) throw new Error('timed out');
|
||||
}
|
||||
Reference in New Issue
Block a user