io-tasks.ts (13647B)
1 // Browser structured lifecycle, independent of tree evaluation and host facilities. 2 // One owning event loop. No promises/callbacks execute inside these transitions. 3 // Host cancellation is a request: release waits for explicit completion acknowledgement. 4 // Resource admission/transfer/reverse release belong to the embedding registry; 5 // this machine is not, by itself, an executable IO interpreter. 6 export interface ResourceGuard { readyForRelease(scope: number): boolean; finishScope(scope: number): void; } 7 export type Kind = 'completed' | 'failed' | 'cancelled'; 8 export interface Diagnostics<T> { readonly items: readonly T[]; readonly omitted: number; } 9 export interface Outcome<T> { readonly kind: Kind; readonly value: T; readonly cleanup?: Diagnostics<T>; } 10 export interface Token { readonly run: bigint; readonly key: number; } 11 export interface HostToken extends Token { readonly scope: number; readonly kind: 'body' | 'release'; } 12 export const hostKey = (token: HostToken): string => `${token.kind}:${token.key}`; 13 type Phase = 'open' | 'joining' | 'releasing'; 14 type ScopeKind = 'root' | 'lexical' | 'spawned'; 15 interface Group<T> { 16 parent: number | null; task: number; kind: ScopeKind; propagate: boolean; 17 phase: Phase; reason: Outcome<T> | null; body: Outcome<T> | null; 18 cleanup: Diagnostics<T>; lexical: number | null; wait: number | null; 19 hostCalls: Set<string>; releaseStarted: boolean; 20 } 21 interface Task<T> { 22 owner: number; supervised: boolean; ticket: boolean; 23 waiter: number | null; outcome: Outcome<T> | null; 24 } 25 const empty = <T>(): Diagnostics<T> => ({ items: [], omitted: 0 }); 26 const primary = <T>(outcome: Outcome<T>): Outcome<T> => ({ kind: outcome.kind, value: outcome.value }); 27 function append<T>(left: Diagnostics<T>, right: Diagnostics<T>): Diagnostics<T> { 28 if (!Number.isSafeInteger(right.omitted) || right.omitted < 0) throw new Error('invalid diagnostics'); 29 const available = 16 - left.items.length; 30 return { items: [...left.items, ...right.items.slice(0, available)], 31 omitted: Math.min(Number.MAX_SAFE_INTEGER, left.omitted + right.omitted + Math.max(0, right.items.length - available)) }; 32 } 33 const capture = <T>(outcome: Outcome<T>): Outcome<T> => { 34 if (!['completed', 'failed', 'cancelled'].includes(outcome.kind)) throw new Error('invalid outcome'); 35 return { ...primary(outcome), cleanup: append(empty(), outcome.cleanup ?? empty()) }; 36 }; 37 export class Tasks<T> { 38 private readonly groups = new Map<number, Group<T>>(); 39 private readonly tasks = new Map<number, Task<T>>(); 40 private readonly hosts = new Map<string, { token: HostToken; cancelling: boolean }>(); 41 private nextScope = 1; 42 private nextHost = 1; 43 constructor(readonly identity: bigint, private readonly ownerStopped: T, private readonly cycle: T, 44 private readonly identityLimit = Number.MAX_SAFE_INTEGER, private readonly resources?: ResourceGuard) { 45 if (typeof identity !== 'bigint' || identity <= 0n || identity > 0xffffffffffffffffn || 46 !Number.isSafeInteger(identityLimit) || identityLimit < 1) throw new Error('invalid run identity or limit'); 47 this.groups.set(0, this.group(null, 0, 'root', false)); 48 } 49 private group(parent: number | null, task: number, kind: ScopeKind, propagate: boolean): Group<T> { 50 return { parent, task, kind, propagate, phase: 'open', reason: null, body: null, 51 cleanup: empty(), lexical: null, wait: null, hostCalls: new Set(), releaseStarted: false }; 52 } 53 private get(scope: number): Group<T> { 54 const group = this.groups.get(scope); 55 if (!group) throw new Error('closed-scope'); 56 return group; 57 } 58 // Read-only copies: embeddings cannot mutate lifecycle state or diagnostic arrays. 59 state(scope: number) { 60 const g = this.get(scope); 61 return { parent: g.parent, task: g.task, kind: g.kind, phase: g.phase, 62 reason: g.reason && primary(g.reason), lexical: g.lexical, wait: g.wait, hostCalls: g.hostCalls.size }; 63 } 64 get counts() { return { scopes: this.groups.size, tasks: this.tasks.size, hosts: this.hosts.size }; } 65 scopeIds(): number[] { return [...this.groups.keys()]; } 66 within(current: number, owner: number): boolean { 67 for (let scope: number | null = current; scope !== null;) { 68 const g = this.groups.get(scope); 69 if (!g) return false; 70 if (scope === owner) return true; 71 scope = g.parent; 72 } 73 return false; 74 } 75 checkpoint(scope: number): void { 76 const g = this.get(scope); 77 if (g.reason) throw new Error('stopped'); 78 if (g.phase !== 'open' || g.lexical !== null || g.wait !== null) throw new Error('invalid-transition'); 79 } 80 private scopeId(): number { 81 if (this.nextScope > this.identityLimit) throw new Error('scope-identity-exhausted'); 82 return this.nextScope++; 83 } 84 nested(parent: number): number { 85 this.checkpoint(parent); 86 const id = this.scopeId(), child = this.group(parent, this.get(parent).task, 'lexical', true); 87 this.groups.set(id, child); this.get(parent).lexical = id; 88 return id; 89 } 90 spawn(owner: number, supervised: boolean, keep: boolean): Token { 91 this.checkpoint(owner); 92 if (typeof supervised !== 'boolean' || typeof keep !== 'boolean') throw new Error('invalid-argument'); 93 const id = this.scopeId(), child = this.group(owner, id, 'spawned', !supervised); 94 this.groups.set(id, child); 95 this.tasks.set(id, { owner, supervised, ticket: keep, waiter: null, outcome: null }); 96 return { run: this.identity, key: id }; 97 } 98 private recordStop(scope: number, reason: Outcome<T>): boolean { 99 const g = this.groups.get(scope); 100 if (!g || g.reason) return false; 101 g.reason = primary(reason); return true; 102 } 103 stop(scope: number, reason: Outcome<T>): void { 104 if (reason.kind !== 'failed' && reason.kind !== 'cancelled') throw new Error('invalid stop'); 105 if (!this.recordStop(scope, reason)) return; 106 // Fixed point propagates first failure upward and owner cancellation downward. 107 for (let changed = true; changed;) { 108 changed = false; 109 for (const [id, g] of this.groups) if (g.parent !== null) { 110 if (g.reason?.kind === 'failed' && g.propagate) changed = this.recordStop(g.parent, g.reason) || changed; 111 if (this.get(g.parent).reason) changed = this.recordStop(id, { kind: 'cancelled', value: this.ownerStopped }) || changed; 112 } 113 } 114 } 115 private checked(current: number, token: Token): Task<T> { 116 if (token.run !== this.identity) throw new Error('stale-task'); 117 const task = this.tasks.get(token.key); 118 if (!task) throw new Error('stale-task'); 119 if (!this.within(current, task.owner)) throw new Error('out-of-scope'); 120 return task; 121 } 122 // Edges are task identities, not lexical scope IDs. Normal scope exit adds 123 // implicit edges to unfinished children; explicit join cycle checks include them. 124 private reaches(start: number, wanted: number): boolean { 125 const pending = [start], seen = new Set<number>(); 126 while (pending.length) { 127 const task = pending.pop()!; 128 if (task === wanted) return true; 129 if (seen.has(task)) continue; 130 seen.add(task); 131 for (const [id, g] of this.groups) if (g.task === task) { 132 if (g.wait !== null) pending.push(g.wait); 133 if (g.phase === 'joining') for (const [child, t] of this.tasks) 134 if (t.owner === id && t.outcome === null) pending.push(child); 135 } 136 } 137 return false; 138 } 139 claimJoin(current: number, token: Token): void { 140 this.checkpoint(current); 141 const task = this.checked(current, token); 142 if (!task.ticket) throw new Error('stale-task'); 143 if (this.within(current, token.key) || this.reaches(token.key, this.get(current).task)) throw new Error('join-cycle'); 144 task.ticket = false; task.waiter = current; this.get(current).wait = token.key; 145 } 146 pollJoin(current: number): Outcome<T> | null { 147 const g = this.get(current); 148 if (g.wait === null) throw new Error('invalid-transition'); 149 const task = this.tasks.get(g.wait)!; 150 if (!task.outcome) return null; 151 const outcome = capture(task.outcome); 152 this.tasks.delete(g.wait); g.wait = null; 153 return outcome; 154 } 155 abandonJoin(current: number): void { 156 const g = this.groups.get(current); 157 if (!g || g.wait === null) return; 158 const task = this.tasks.get(g.wait)!; 159 task.waiter = null; 160 if (task.outcome) this.tasks.delete(g.wait); 161 g.wait = null; 162 } 163 discard(current: number, token: Token): void { 164 this.checkpoint(current); 165 const task = this.checked(current, token); 166 if (!task.ticket) throw new Error('stale-task'); 167 task.ticket = false; 168 if (task.outcome) this.tasks.delete(token.key); 169 } 170 cancel(current: number, token: Token, reason: T): void { 171 this.checkpoint(current); 172 const task = this.checked(current, token); 173 if (task.outcome) throw new Error('stale-task'); 174 this.stop(token.key, { kind: 'cancelled', value: reason }); 175 } 176 bodyDone(scope: number, outcome: Outcome<T>): void { 177 const g = this.get(scope), captured = capture(outcome); 178 if (g.phase !== 'open' || g.lexical !== null) throw new Error('invalid-transition'); 179 this.abandonJoin(scope); 180 if (captured.kind !== 'completed') this.stop(scope, captured); 181 g.body = captured; 182 for (const [id, t] of this.tasks) if (t.owner === scope && !t.outcome && this.reaches(id, g.task)) { 183 this.stop(scope, { kind: 'failed', value: this.cycle }); break; 184 } 185 g.phase = 'joining'; 186 } 187 beginHost(scope: number): HostToken { 188 this.checkpoint(scope); 189 return this.registerHost(scope, 'body'); 190 } 191 beginReleaseHost(scope: number): HostToken { 192 const group = this.get(scope); 193 if (group.phase !== 'releasing' || group.releaseStarted) throw new Error('invalid-transition'); 194 // One cleanup batch per scope uses its already-reserved identity, in a 195 // separate namespace. Body-call identity exhaustion cannot prevent cleanup. 196 const token = this.registerHost(scope, 'release'); 197 group.releaseStarted = true; 198 return token; 199 } 200 private registerHost(scope: number, kind: HostToken['kind']): HostToken { 201 if (kind === 'body' && this.nextHost > this.identityLimit) throw new Error('host-identity-exhausted'); 202 const token = Object.freeze({ run: this.identity, key: kind === 'body' ? this.nextHost++ : scope, scope, kind }); 203 const key = hostKey(token); 204 this.hosts.set(key, { token, cancelling: false }); this.get(scope).hostCalls.add(key); 205 return token; 206 } 207 // Delivery is separate from acknowledgement. Issuing AbortController.abort() 208 // cannot settle a task whose host callback/lease is still outstanding. 209 takeCancellations(): HostToken[] { 210 const pending: HostToken[] = []; 211 for (const call of this.hosts.values()) if (call.token.kind === 'body' && !call.cancelling && this.get(call.token.scope).reason) { 212 call.cancelling = true; pending.push(call.token); 213 } 214 return pending; 215 } 216 endHost(token: HostToken): 'reply' | 'ignored' | 'stale' { 217 const key = hostKey(token), call = this.hosts.get(key); 218 if (token.run !== this.identity || !call || call.token.scope !== token.scope || call.token.kind !== token.kind) return 'stale'; 219 const g = this.get(token.scope); 220 this.hosts.delete(key); g.hostCalls.delete(key); 221 return token.kind === 'release' || !g.reason && !g.body ? 'reply' : 'ignored'; 222 } 223 beginRelease(scope: number): boolean { 224 const g = this.get(scope); 225 if (g.phase !== 'joining') throw new Error('invalid-transition'); 226 if (g.hostCalls.size || [...this.tasks.values()].some(t => t.owner === scope && !t.outcome) || 227 this.resources && !this.resources.readyForRelease(scope)) return false; 228 g.phase = 'releasing'; return true; 229 } 230 // Call only after the embedding has physically released this scope's resources. 231 settle(scope: number, ownCleanup: Diagnostics<T> = empty()): Outcome<T> { 232 const g = this.get(scope); 233 if (g.phase !== 'releasing' || !g.body || g.hostCalls.size) throw new Error('invalid-transition'); 234 let final = primary(g.reason ?? g.body); 235 let errors = append<T>(append<T>(g.body.cleanup ?? empty<T>(), g.cleanup), ownCleanup); 236 if (final.kind === 'completed' && errors.items.length) { 237 final = { kind: 'failed', value: errors.items[0]! }; 238 errors = { items: errors.items.slice(1), omitted: errors.omitted }; 239 } 240 this.resources?.finishScope(scope); 241 final = { ...final, cleanup: errors }; 242 if (final.kind === 'failed') this.stop(scope, final); 243 for (const [id, task] of this.tasks) if (task.owner === scope) this.tasks.delete(id); 244 if (g.parent !== null) { 245 const parent = this.get(g.parent); 246 if (g.kind === 'lexical') { 247 parent.cleanup = append(parent.cleanup, errors); parent.lexical = null; 248 } else { 249 const task = this.tasks.get(scope)!; 250 if (!task.supervised) parent.cleanup = append(parent.cleanup, errors); 251 task.outcome = final; 252 if (!task.ticket && task.waiter === null) this.tasks.delete(scope); 253 } 254 } 255 this.groups.delete(scope); 256 return g.kind === 'lexical' ? primary(final) : capture(final); 257 } 258 // Evaluator roots, in addition to embedding-owned bodies, continuations and 259 // pending host arguments. No task/result history remains after consumption. 260 *payloads(): Iterable<T> { 261 yield this.ownerStopped; yield this.cycle; 262 for (const g of this.groups.values()) { 263 if (g.reason) yield g.reason.value; 264 if (g.body) { yield g.body.value; yield* g.body.cleanup?.items ?? []; } 265 yield* g.cleanup.items; 266 } 267 for (const task of this.tasks.values()) if (task.outcome) { 268 yield task.outcome.value; yield* task.outcome.cleanup?.items ?? []; 269 } 270 } 271 }