io-host-work.js (4095B)
1 import { hostKey } from './io-tasks.js'; 2 // Adapter-side asynchronous work. Starting one callback never waits for it, so 3 // sibling actions/cancellation can proceed. The provider must settle its promise 4 // AFTER cancellation cleanup; abort delivery alone is never an acknowledgement. 5 export class HostWork { 6 tasks; 7 active = new Map(); 8 replies = new Map(); 9 notification = null; 10 constructor(tasks) { 11 this.tasks = tasks; 12 } 13 start(scope, operation) { 14 return this.register(this.tasks.beginHost(scope), operation); 15 } 16 // Cleanup is not cancelled by a new stop request. Its result (including a 17 // failure) must still reach the driver to preserve primary/secondary outcomes. 18 release(scope, operation) { 19 return this.register(this.tasks.beginReleaseHost(scope), operation); 20 } 21 register(token, operation) { 22 const controller = new AbortController(); 23 this.active.set(hostKey(token), { token, controller }); 24 const complete = (outcome) => { 25 // Settle first; a cancelled reply is discarded without observing its value. 26 const disposition = this.tasks.endHost(token); 27 this.active.delete(hostKey(token)); 28 if (disposition === 'reply') 29 this.replies.set(hostKey(token), { token, outcome }); 30 const notice = this.notification; 31 this.notification = null; 32 notice?.resolve(); 33 }; 34 try { 35 Promise.resolve(operation(controller.signal)).then(value => complete({ ok: true, value }), error => complete({ ok: false, error })); 36 } 37 catch (error) { 38 complete({ ok: false, error }); 39 } 40 return token; 41 } 42 usable(token) { 43 if (!this.tasks.within(token.scope, token.scope)) 44 return false; 45 const state = this.tasks.state(token.scope); 46 return token.kind === 'release' ? state.phase === 'releasing' : state.phase === 'open' && state.reason === null; 47 } 48 // Resource close may interrupt one operation without cancelling its task. 49 // The resulting provider reply still reaches that body as an ordinary error. 50 // Never abort a cleanup batch or acknowledge work merely by sending this signal. 51 cancelBody(token) { 52 const work = this.active.get(hostKey(token)); 53 if (token.kind !== 'body' || !work || work.token.run !== token.run || 54 work.token.scope !== token.scope || work.controller.signal.aborted) 55 return false; 56 work.controller.abort(); 57 return true; 58 } 59 cancelStopped() { 60 for (const token of this.tasks.takeCancellations()) { 61 const work = this.active.get(hostKey(token)); 62 if (!work || work.token.scope !== token.scope) 63 throw new Error('host work ownership mismatch'); 64 work.controller.abort(); 65 } 66 // Completion may have won the race with cancellation. Drop undelivered data 67 // too; a queued reply cannot revive a stopped/settled body. 68 for (const [key, reply] of this.replies) 69 if (!this.usable(reply.token)) 70 this.replies.delete(key); 71 } 72 take() { 73 for (const [key, reply] of this.replies) { 74 this.replies.delete(key); 75 if (this.usable(reply.token)) 76 return reply; 77 } 78 return null; 79 } 80 wait() { 81 if (this.replies.size || !this.active.size) 82 return Promise.resolve(); 83 if (!this.notification) { 84 let resolve; 85 const promise = new Promise(done => { resolve = done; }); 86 this.notification = { promise, resolve }; 87 } 88 return this.notification.promise; 89 } 90 get counts() { return { pending: this.active.size, replies: this.replies.size }; } 91 // Reply data roots are the embedding's responsibility until taken/discarded. 92 *values() { 93 for (const reply of this.replies.values()) 94 if (this.usable(reply.token) && reply.outcome.ok) 95 yield reply.outcome.value; 96 } 97 }