io-v2-core.js (15059B)
1 // Private resumable tree interpreter. No public session dispatch or capabilities. 2 // Adapters must retain roots() between turns and keep temporary evaluator handles 3 // alive for a whole turn. Provider promises acknowledge physical completion; 4 // providers return deferred Result encoders, never task outcomes. Encoding runs 5 // in an interpreter turn, not in a promise/GC gap; cancelled encoders never run. 6 import { canonicalNat, decodeAction, ProtocolError } from './io-v2-protocol.js'; 7 import { lifecycle } from './io-v2-resources.js'; 8 import { HostWork } from './io-v2-host-work.js'; 9 const operationErrors = new Set(['stale-task', 'join-cycle', 'out-of-scope', 'stale-handle', 'not-owner', 10 'borrowed-resource', 'internal-resource', 'closed-handle', 'busy', 'duplicate-handle', 'closed-scope', 'invalid-argument']); 11 export class OperationError extends Error { 12 } 13 function logical(operation) { 14 try { 15 return operation(); 16 } 17 catch (error) { 18 if (error instanceof Error && operationErrors.has(error.message)) 19 throw new OperationError(error.message); 20 throw error; 21 } 22 } 23 export class Wire { 24 tree; 25 identity; 26 constructor(tree, identity) { 27 this.tree = tree; 28 this.identity = identity; 29 } 30 pair(value) { 31 const node = this.tree.observe(value).node; 32 if (node.kind !== 'fork') 33 throw new OperationError('invalid-argument'); 34 return [node.left, node.right]; 35 } 36 list(values) { let out = this.tree.leaf; for (let i = values.length - 1; i >= 0; i--) 37 out = this.tree.fork(values[i], out); return out; } 38 result(ok, value) { return this.tree.fork(ok ? this.tree.stem(this.tree.leaf) : this.tree.leaf, this.tree.fork(value, this.tree.leaf)); } 39 problem(category, operation, detail = '', progress = 'none') { 40 return this.list([category, operation, detail, progress].map(s => this.tree.text(s))); 41 } 42 marker(name, key) { return this.list([this.tree.text(name), this.tree.natural(this.identity), this.tree.natural(BigInt(key))]); } 43 token(value, name) { 44 const [tag, rest] = this.pair(value), [run, tail] = this.pair(rest), [key, end] = this.pair(tail); 45 if (this.tree.observe(end).node.kind !== 'leaf') 46 throw new OperationError('invalid-argument'); 47 let cursor = tag; 48 for (const char of name) { 49 const [head, next] = this.pair(cursor); 50 if (canonicalNat(this.tree, head, 7) !== BigInt(char.charCodeAt(0))) 51 throw new OperationError('invalid-argument'); 52 cursor = next; 53 } 54 if (this.tree.observe(cursor).node.kind !== 'leaf') 55 throw new OperationError('invalid-argument'); 56 const r = canonicalNat(this.tree, run, 64), k = canonicalNat(this.tree, key, 64); 57 if (r === undefined || k === undefined) 58 throw new OperationError('invalid-argument'); 59 // Unrepresentable keys cannot alias a live JS identity. Keep the ordinary 60 // stale-handle/task path rather than rounding into a different live handle. 61 return { run: r, key: k > BigInt(Number.MAX_SAFE_INTEGER) ? -1 : Number(k) }; 62 } 63 outcome(value) { 64 const name = this.tree.text(value.kind); 65 if (value.kind === 'completed') 66 return this.list([name, value.value]); 67 const errors = [...value.cleanup?.items ?? []], omitted = value.cleanup?.omitted ?? 0; 68 if (omitted) 69 errors.push(this.problem('cleanup-omitted', 'close', String(omitted))); 70 return this.list([name, value.value, this.list(errors)]); 71 } 72 } 73 export class Core { 74 tree; 75 provider; 76 actionLimit; 77 wire; 78 lifecycle; 79 work; 80 bodies = new Map(); 81 constants; 82 pendingArguments = new Map(); 83 cursor = -1; 84 actions = 0; 85 result = null; 86 // Embedding/evaluator failure is retained separately and never stringified or 87 // mistaken for an ordinary Result. Drain cleanup before rethrowing at boundary. 88 fault = undefined; 89 hasFault = false; 90 constructor(tree, identity, action, standard, provider, actionLimit = 65536) { 91 this.tree = tree; 92 this.provider = provider; 93 this.actionLimit = actionLimit; 94 if (!Number.isSafeInteger(actionLimit) || actionLimit < 1) 95 throw new Error('invalid action limit'); 96 this.wire = new Wire(tree, identity); 97 this.constants = { 98 stopped: this.wire.problem('cancelled', 'scope', 'owner stopped'), 99 cycle: this.wire.problem('join-cycle', 'scope'), fault: this.wire.problem('runtime-error', 'core'), 100 cleanup: this.wire.problem('cleanup-failed', 'resource/release', '', 'unknown'), budget: this.wire.problem('limit', 'core', 'IO action limit exceeded') 101 }; 102 this.lifecycle = lifecycle(identity, this.constants.stopped, this.constants.cycle, standard); 103 this.work = new HostWork(this.lifecycle.tasks); 104 this.bodies.set(0, this.body(action)); 105 } 106 body(action) { return { action, frames: [], mode: 'runnable' }; } 107 stop(reason) { if (!this.result) { 108 this.lifecycle.tasks.stop(0, { kind: 'cancelled', value: reason }); 109 this.work.cancelStopped(); 110 } } 111 done(scope, outcome) { 112 this.lifecycle.tasks.bodyDone(scope, outcome); 113 const body = this.bodies.get(scope); 114 body.mode = 'done'; 115 body.action = this.tree.leaf; 116 body.frames = []; 117 } 118 value(scope, value) { 119 const body = this.bodies.get(scope); 120 if (body.frames.length) { 121 body.action = this.tree.app(body.frames.pop(), value); 122 body.mode = 'runnable'; 123 } 124 else 125 this.done(scope, { kind: 'completed', value }); 126 } 127 operation(scope, name, action) { 128 try { 129 action(); 130 } 131 catch (error) { 132 if (!(error instanceof OperationError)) 133 throw error; 134 this.value(scope, this.wire.result(false, this.wire.problem(error.message, name))); 135 } 136 } 137 step(scope) { 138 const { tasks, resources } = this.lifecycle, body = this.bodies.get(scope), state = tasks.state(scope); 139 if (body.mode === 'done' || body.mode === 'lexical') 140 return false; 141 if (state.reason) { 142 this.done(scope, state.reason); 143 return true; 144 } 145 if (body.mode === 'host') 146 return false; 147 if (body.mode === 'join') { 148 const outcome = tasks.pollJoin(scope); 149 if (!outcome) 150 return false; 151 this.value(scope, this.wire.result(true, this.wire.outcome(outcome))); 152 return true; 153 } 154 if (++this.actions > this.actionLimit) { 155 tasks.stop(0, { kind: 'failed', value: this.constants.budget }); 156 return true; 157 } 158 let action; 159 try { 160 action = decodeAction(this.tree, body.action); 161 } 162 catch (error) { 163 if (!(error instanceof ProtocolError)) 164 throw error; 165 this.done(scope, { kind: 'failed', value: this.wire.problem('protocol-error', 'core/decode', error.message) }); 166 return true; 167 } 168 if (action.kind === 'request') { 169 // Keep action/arguments rooted until provider acknowledgement, including 170 // cancellation. The abandoned continuation is cleared by done(). 171 body.mode = 'host'; 172 const request = action; 173 this.pendingArguments.set(scope, request.arguments); 174 this.work.start(scope, async (signal) => ({ kind: 'body', encode: await this.provider.request(this, scope, request, signal) })); 175 return true; 176 } 177 const p = action.payload; 178 switch (action.code) { 179 case 0: 180 this.value(scope, p); 181 break; 182 case 1: { 183 const node = this.tree.observe(p).node; 184 if (node.kind !== 'fork') 185 this.done(scope, { kind: 'failed', value: this.wire.problem('protocol-error', 'core/decode', 'invalid v2 computation') }); 186 else { 187 body.frames.push(node.right); 188 body.action = node.left; 189 } 190 break; 191 } 192 case 2: 193 this.done(scope, { kind: 'failed', value: p }); 194 break; 195 case 9: 196 this.done(scope, { kind: 'cancelled', value: p }); 197 break; 198 case 3: { 199 const child = tasks.nested(scope); 200 this.bodies.set(child, this.body(this.tree.leaf)); 201 body.mode = 'lexical'; 202 body.action = this.tree.leaf; 203 this.bodies.get(child).action = this.tree.app(p, this.wire.marker('io.v2.scope', child)); 204 break; 205 } 206 case 4: 207 this.value(scope, this.wire.marker('io.v2.scope', scope)); 208 break; 209 case 5: 210 this.operation(scope, 'core/spawn', () => { 211 const [sup, a] = this.wire.pair(p), [keep, b] = this.wire.pair(a), [handles, callback] = this.wire.pair(b); 212 const supervised = canonicalNat(this.tree, sup, 1), retained = canonicalNat(this.tree, keep, 1); 213 if (supervised === undefined || retained === undefined) 214 throw new OperationError('invalid-argument'); 215 const tokens = []; 216 let rest = handles; 217 while (this.tree.observe(rest).node.kind !== 'leaf') { 218 if (tokens.length === 256) 219 throw new OperationError('invalid-argument'); 220 const [head, tail] = this.wire.pair(rest); 221 tokens.push(this.wire.token(head, 'io.v2.resource')); 222 rest = tail; 223 } 224 const moved = logical(() => resources.spawn(scope, supervised === 1n, retained === 1n, tokens)), child = moved.task.key; 225 this.bodies.set(child, this.body(this.tree.leaf)); 226 this.bodies.get(child).action = this.tree.app(callback, this.wire.list(moved.handles.map(t => this.wire.marker('io.v2.resource', t.key)))); 227 this.value(scope, this.wire.result(true, this.wire.list([this.wire.marker('io.v2.task', child), retained === 1n ? this.wire.marker('io.v2.join', child) : this.tree.leaf]))); 228 }); 229 break; 230 case 6: 231 case 7: 232 this.operation(scope, action.code === 6 ? 'core/join' : 'core/discard', () => { 233 const token = this.wire.token(p, 'io.v2.join'); 234 if (action.code === 7) { 235 logical(() => tasks.discard(scope, token)); 236 this.value(scope, this.wire.result(true, this.tree.leaf)); 237 } 238 else { 239 logical(() => tasks.claimJoin(scope, token)); 240 body.mode = 'join'; 241 body.action = this.tree.leaf; 242 } 243 }); 244 break; 245 case 8: 246 this.operation(scope, 'core/cancel', () => { 247 const [handle, reason] = this.wire.pair(p), token = this.wire.token(handle, 'io.v2.task'); 248 logical(() => tasks.cancel(scope, token, reason)); 249 this.value(scope, this.wire.result(true, this.tree.leaf)); 250 }); 251 break; 252 } 253 return true; 254 } 255 settled(scope, errors) { 256 const { tasks } = this.lifecycle, state = tasks.state(scope), outcome = tasks.settle(scope, errors); 257 this.bodies.delete(scope); 258 if (scope === 0) 259 this.result = outcome; 260 else if (state.kind === 'lexical') { 261 const parent = state.parent, reason = tasks.state(parent).reason; 262 this.bodies.get(parent).mode = 'runnable'; 263 if (reason) 264 this.done(parent, reason); 265 else if (outcome.kind === 'completed') 266 this.value(parent, outcome.value); 267 else 268 this.done(parent, outcome); 269 } 270 } 271 // One progress turn, never waits for a provider. False means wait for host 272 // acknowledgement. The embedding must yield to its event loop between batches. 273 advance() { 274 if (this.result) 275 return false; 276 try { 277 this.work.cancelStopped(); 278 for (const scope of this.pendingArguments.keys()) 279 if (this.lifecycle.tasks.state(scope).hostCalls === 0) 280 this.pendingArguments.delete(scope); 281 const reply = this.work.take(); 282 if (reply) { 283 if (!reply.outcome.ok) 284 throw reply.outcome.error; 285 const value = reply.outcome.value; 286 if (value.kind === 'release') 287 this.settled(reply.token.scope, value.errors); 288 else 289 this.value(reply.token.scope, value.encode()); 290 return true; 291 } 292 const { tasks, resources } = this.lifecycle; 293 for (const scope of tasks.scopeIds()) 294 if (tasks.state(scope).phase === 'joining' && tasks.beginRelease(scope)) { 295 this.work.release(scope, async () => ({ kind: 'release', errors: await resources.releaseScope(scope, resource => this.provider.release(resource), () => this.constants.cleanup, this.constants.cleanup) })); 296 return true; 297 } 298 const ids = tasks.scopeIds(), ordered = [...ids.filter(id => id > this.cursor), ...ids.filter(id => id <= this.cursor)]; 299 for (const scope of ordered) 300 if (this.step(scope)) { 301 this.cursor = scope; 302 return true; 303 } 304 return false; 305 } 306 catch (error) { 307 if (!this.hasFault) 308 this.fault = error; 309 this.hasFault = true; 310 this.lifecycle.tasks.stop(0, { kind: 'failed', value: this.constants.fault }); 311 this.work.cancelStopped(); 312 return true; 313 } 314 } 315 wait() { return this.work.wait(); } 316 *roots() { 317 yield this.tree.leaf; 318 yield* Object.values(this.constants); 319 yield* this.lifecycle.tasks.payloads(); 320 yield* this.pendingArguments.values(); 321 yield* this.provider.roots?.() ?? []; 322 for (const body of this.bodies.values()) { 323 yield body.action; 324 yield* body.frames; 325 } 326 for (const value of this.work.values()) { 327 if (value.kind === 'release') 328 yield* value.errors.items; 329 } 330 if (this.result) { 331 yield this.result.value; 332 yield* this.result.cleanup?.items ?? []; 333 } 334 } 335 }