185 lines
11 KiB
TypeScript
185 lines
11 KiB
TypeScript
import type { Id, Json, OperationIntent } from '../canvas-core/types.ts';
|
|
import { fault, readMap } from '../canvas-core/values.ts';
|
|
import { createLedger, type FrozenRequest, type LedgerSnapshot, type Outcome } from './ledger.ts';
|
|
import { createDeque, createMinHeap } from './structures.ts';
|
|
|
|
export type OperationStatus = { readonly outcome: Outcome; readonly error?: string; readonly retryAfterMs?: number; readonly canonical?: Json };
|
|
export type CompiledRequest = { readonly payload: Json; readonly canvasIds: readonly Id[]; readonly expectedVersions: Readonly<Record<Id, number>> };
|
|
export type Clock = { now(): number; setTimer(callback: () => void, delay: number): unknown; clearTimer(handle: unknown): void };
|
|
const browserClock: Clock = { now: () => Date.now(), setTimer: (fn, delay) => setTimeout(fn, delay), clearTimer: handle => clearTimeout(handle as ReturnType<typeof setTimeout>) };
|
|
type Ledger = ReturnType<typeof createLedger>;
|
|
export type SchedulerOptions = {
|
|
getLedger(): LedgerSnapshot;
|
|
mutate(reduce: (ledger: Ledger) => void): void;
|
|
persist(): Promise<void>;
|
|
createId(): Id;
|
|
participants(intent: OperationIntent): readonly Id[];
|
|
compile(intent: OperationIntent, requestId: Id): CompiledRequest;
|
|
submit(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
|
|
query(request: FrozenRequest, signal: AbortSignal): Promise<OperationStatus>;
|
|
supportsRequest?: (request: FrozenRequest) => boolean;
|
|
/** Atomically reconcile the receipt/base/working document and the ledger outcome. */
|
|
complete(request: FrozenRequest, status: OperationStatus): { readonly blocked: true } | void;
|
|
onError(error: unknown): void;
|
|
clock?: Clock;
|
|
concurrency?: number;
|
|
retryDelay?: (attempt: number) => number;
|
|
};
|
|
|
|
/** Queue/heap/claims are indexes only. Every authoritative transition lives in the ledger. */
|
|
export function createScheduler(options: SchedulerOptions) {
|
|
const clock = options.clock ?? browserClock, concurrency = options.concurrency ?? 2;
|
|
if (!Number.isSafeInteger(concurrency) || concurrency < 1) throw fault('invalid_concurrency');
|
|
const ready = createDeque<Id>(), queued = new Set<Id>(), claims = new Map<Id, Id>();
|
|
const waiters = new Map<Id, Set<Id>>(), dependents = new Map<Id, Set<Id>>(), indexed = new Set<Id>();
|
|
const retry = createMinHeap<{ id: Id; dueAt: number; sequence: number; token: number }>((a, b) => a.dueAt - b.dueAt || a.sequence - b.sequence);
|
|
const tokens = new Map<Id, number>(), active = new Map<Id, { controller: AbortController; promise: Promise<void> }>();
|
|
let timer: unknown, enqueueSequence = 0, stopped = true, closed = false, pumping = false;
|
|
const delay = (attempt: number) => Math.max(1, options.retryDelay?.(attempt) ?? Math.min(30_000, 500 * 2 ** Math.min(attempt, 6)));
|
|
const report = (error: unknown) => { try { options.onError(error); } catch { /* Reporting cannot change a receipt. */ } };
|
|
const supported = (request: FrozenRequest) => options.supportsRequest?.(request) ?? request.protocolVersion === 1;
|
|
|
|
function enqueue(id: Id) {
|
|
if (queued.has(id)) return;
|
|
const record = options.getLedger().operations.get(id);
|
|
if (!record || record.phase !== 'pending' || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) return;
|
|
queued.add(id); ready.push(id);
|
|
}
|
|
function index(id: Id) {
|
|
const record = options.getLedger().operations.get(id);
|
|
if (!record) return;
|
|
if (!indexed.has(id)) {
|
|
indexed.add(id);
|
|
for (const dependency of record.intent.dependencies) {
|
|
const set = dependents.get(dependency) ?? new Set<Id>(); set.add(id); dependents.set(dependency, set);
|
|
}
|
|
}
|
|
enqueue(id);
|
|
}
|
|
function own(request: FrozenRequest) {
|
|
for (const canvas of request.canvasIds) {
|
|
const prior = claims.get(canvas);
|
|
if (prior && prior !== request.id) throw fault('recovery_claim_conflict', [canvas]);
|
|
}
|
|
for (const canvas of request.canvasIds) claims.set(canvas, request.id);
|
|
}
|
|
function release(request: FrozenRequest) {
|
|
for (const canvas of request.canvasIds) if (claims.get(canvas) === request.id) {
|
|
claims.delete(canvas);
|
|
const waiting = waiters.get(canvas); waiters.delete(canvas);
|
|
for (const id of waiting ?? []) enqueue(id);
|
|
}
|
|
for (const id of request.operationIds) for (const dependent of dependents.get(id) ?? []) enqueue(dependent);
|
|
}
|
|
function schedule(id: Id, dueAt: number, action: 'submit' | 'query', persist = true) {
|
|
if (persist) options.mutate(ledger => ledger.scheduleRetry(id, dueAt, action));
|
|
const token = (tokens.get(id) ?? 0) + 1; tokens.set(id, token);
|
|
retry.push({ id, dueAt, sequence: ++enqueueSequence, token }); armTimer();
|
|
}
|
|
function validRetry(item: { id: Id; token: number }) {
|
|
return tokens.get(item.id) === item.token && options.getLedger().requests.get(item.id)?.phase !== 'resolved';
|
|
}
|
|
function armTimer() {
|
|
if (timer !== undefined) { clock.clearTimer(timer); timer = undefined; }
|
|
while (retry.peek() && !validRetry(retry.peek()!)) retry.pop();
|
|
if (stopped || !retry.peek()) return;
|
|
timer = clock.setTimer(() => { timer = undefined; pump(); }, Math.max(0, retry.peek()!.dueAt - clock.now()));
|
|
}
|
|
function launch(id: Id, action: 'submit' | 'query') {
|
|
if (active.has(id) || stopped) return;
|
|
const controller = new AbortController();
|
|
// Start in a microtask so the handle exists before any synchronous adapter error.
|
|
const promise = Promise.resolve().then(async () => {
|
|
try {
|
|
if (stopped) return;
|
|
if (action === 'submit') options.mutate(ledger => ledger.markDispatch(id));
|
|
// This barrier includes the exact body/key and the conservative dispatch marker.
|
|
await options.persist();
|
|
if (stopped) return;
|
|
const request = options.getLedger().requests.get(id)!;
|
|
let status: OperationStatus;
|
|
try { status = await (action === 'submit' ? options.submit(request, controller.signal) : options.query(request, controller.signal)); }
|
|
catch { status = { outcome: 'unresolved', error: 'transport_unknown' }; }
|
|
if (stopped) return; // A stale identity/runtime cannot accept a late callback.
|
|
const completion = options.complete(request, status);
|
|
const current = options.getLedger().requests.get(id)!;
|
|
if (current.phase === 'resolved') { tokens.delete(id); await options.persist(); if (!stopped) release(current); }
|
|
else if (completion?.blocked) { await options.persist(); }
|
|
else {
|
|
// A missing receipt is unknown, so the only write retry reuses this frozen request.
|
|
const nextAction = status.outcome === 'accepted' || action === 'submit' ? 'query' : 'submit';
|
|
schedule(id, clock.now() + Math.max(1, status.retryAfterMs ?? delay(current.attempt)), nextAction);
|
|
await options.persist();
|
|
}
|
|
} catch (error) { report(error); } // Storage/reconciliation failure keeps its claims held.
|
|
}).finally(() => { active.delete(id); if (!stopped) pump(); });
|
|
active.set(id, { controller, promise });
|
|
}
|
|
function pump() {
|
|
if (stopped || pumping) return;
|
|
pumping = true;
|
|
try {
|
|
while (active.size < concurrency && retry.peek() && retry.peek()!.dueAt <= clock.now()) {
|
|
const item = retry.pop()!;
|
|
if (!validRetry(item)) continue;
|
|
const request = options.getLedger().requests.get(item.id);
|
|
if (request && !active.has(item.id)) launch(item.id, request.retryAction ?? 'query');
|
|
}
|
|
let budget = ready.size;
|
|
while (active.size < concurrency && budget-- > 0) {
|
|
const id = ready.pop(); if (!id) break;
|
|
queued.delete(id);
|
|
const record = options.getLedger().operations.get(id);
|
|
if (!record || record.phase !== 'pending' || record.blockers.length || record.intent.dependencies.some(dependency => options.getLedger().operations.get(dependency)?.phase !== 'resolved')) continue;
|
|
try {
|
|
const canvases = [...new Set(options.participants(record.intent))].sort();
|
|
if (!canvases.length) throw fault('missing_participants');
|
|
const occupied = canvases.filter(canvas => claims.has(canvas));
|
|
if (occupied.length) {
|
|
for (const canvas of occupied) { const set = waiters.get(canvas) ?? new Set<Id>(); set.add(id); waiters.set(canvas, set); }
|
|
continue;
|
|
}
|
|
const requestId = options.createId(), compiled = options.compile(record.intent, requestId);
|
|
if (canvases.length !== compiled.canvasIds.length || canvases.some(canvas => !compiled.canvasIds.includes(canvas))) throw fault('participant_mismatch');
|
|
options.mutate(ledger => ledger.freeze({ id: requestId, operationIds: [id], ...compiled }));
|
|
own(options.getLedger().requests.get(requestId)!); launch(requestId, 'submit');
|
|
} catch (error) { options.mutate(ledger => ledger.block(id, [(error as Error).message || 'compile_failed'])); report(error); }
|
|
}
|
|
} finally { pumping = false; armTimer(); }
|
|
}
|
|
return Object.freeze({
|
|
start() {
|
|
if (closed) throw fault('scheduler_closed');
|
|
if (!stopped) return;
|
|
// Validate the entire recovered claim set before sending anything.
|
|
claims.clear();
|
|
for (const request of options.getLedger().requests.values()) if (request.phase !== 'resolved') own(request);
|
|
stopped = false;
|
|
for (const operation of [...options.getLedger().operations.values()].sort((a, b) => a.intent.sequence - b.intent.sequence)) index(operation.intent.id);
|
|
for (const request of options.getLedger().requests.values()) if (request.phase !== 'resolved') {
|
|
if (!supported(request)) { options.mutate(ledger => ledger.blockRequest(request.id, 'request_interpreter_unavailable')); continue; }
|
|
schedule(request.id, request.retryAt ?? clock.now(), request.retryAction ?? (request.phase === 'frozen' ? 'submit' : 'query'), false);
|
|
}
|
|
pump();
|
|
},
|
|
notify(operationIds: readonly Id[]) { for (const id of operationIds) index(id); pump(); },
|
|
/** Explicit retry after storage/auth/permission recovery; preserves frozen bodies. */
|
|
async resume() {
|
|
if (stopped) return;
|
|
// A receipt may be in memory after its durable write failed. Keep claims until retry succeeds.
|
|
await options.persist();
|
|
if (stopped) return;
|
|
for (const request of options.getLedger().requests.values()) {
|
|
if (active.has(request.id)) continue;
|
|
if (request.phase === 'resolved') release(request);
|
|
else if (supported(request)) schedule(request.id, request.retryAt ?? clock.now(), request.retryAction ?? 'query', false);
|
|
}
|
|
for (const id of indexed) enqueue(id);
|
|
pump();
|
|
},
|
|
getClaims: () => readMap(claims),
|
|
async settle() { while (active.size) await Promise.all([...active.values()].map(item => item.promise)); },
|
|
stop() { closed = true; stopped = true; if (timer !== undefined) clock.clearTimer(timer); timer = undefined; for (const item of active.values()) item.controller.abort(); },
|
|
});
|
|
}
|