also looking at this
feat: add bounded VTOL wind braking
#88
13 files
+447
-25
Review threads live on the whole diff, not on one commit, so none are shown here - a thread's line means something in the branch's final form, and painting it into an intermediate step would put it on code it is not about.
| @@ -513,6 +513,7 @@ jobs: | ||
| 513 | 513 | ${{ runner.temp }}/execution-browser-evidence/admission-renewal.json |
| 514 | 514 | ${{ runner.temp }}/execution-browser-evidence/pre-admission-renewal.json |
| 515 | 515 | ${{ runner.temp }}/execution-browser-evidence/completion-read-renewal.json |
| 516 | ${{ runner.temp }}/execution-browser-evidence/worker-callback-timings.json | |
| 516 | 517 | ${{ runner.temp }}/execution-browser-evidence/chrome-startup-stderr.log |
| 517 | 518 | purge-namespace: |
| 518 | 519 | runs-on: ubuntu-latest |
| @@ -5,6 +5,7 @@ import { mkdir, writeFile } from 'node:fs/promises' | ||
| 5 | 5 | import { join, resolve } from 'node:path' |
| 6 | 6 | import { randomUUID } from 'node:crypto' |
| 7 | 7 | import { setupHandoverFixture } from './handover-setup' |
| 8 | import { createExecutionCallbackObserver } from '../../helpers/execution-worker-observation' | |
| 8 | 9 | |
| 9 | 10 | // Real native setup owns the guarded disposable SQLite file before any application boot. |
| 10 | 11 | const fixture = await setupHandoverFixture() |
| @@ -43,6 +44,14 @@ let nextOrdinal = 0 | ||
| 43 | 44 | const scalar = (value: unknown) => typeof value === 'number' && Number.isFinite(value) && Math.abs(value) <= Number.MAX_SAFE_INTEGER ? value : null |
| 44 | 45 | const field = (value: unknown, name: string): unknown => value && typeof value === 'object' ? Object.getOwnPropertyDescriptor(value, name)?.value : undefined |
| 45 | 46 | const choice = <T extends string>(value: unknown, values: readonly T[]): T | null => values.find(item => item === value) ?? null |
| 47 | const callbackObserver = createExecutionCallbackObserver(value => console.log(JSON.stringify(value))) | |
| 48 | function observeAcceptedWorker(acceptedOrdinal: number) { | |
| 49 | try { | |
| 50 | const worker = field(executionOwner().slot, 'worker') | |
| 51 | if (worker instanceof Worker) callbackObserver.observe(worker, acceptedOrdinal) | |
| 52 | } | |
| 53 | catch { /* Private observation must not change an accepted native response. */ } | |
| 54 | } | |
| 46 | 55 | const ordinal = (slot: object) => { |
| 47 | 56 | const existing = ordinals.get(slot) |
| 48 | 57 | if (existing !== undefined) return existing |
| @@ -61,7 +70,7 @@ function diagnosticSnapshot() { | ||
| 61 | 70 | return typeof value === 'number' ? scalar(value - now) : null |
| 62 | 71 | } |
| 63 | 72 | return { |
| 64 | available: true, pendingChecks: scalar(owner.pendingChecks), | |
| 73 | available: true, pendingChecks: scalar(owner.pendingChecks), workerCallbacks: callbackObserver.snapshot(), | |
| 65 | 74 | fixture: { startRequests, admissionRequests, streamStarts, holdArmed, held, holdUsed, holdWorkspace, workspaceHeld, preAdmissionRenewalArmed, preAdmissionRenewed, preAdmissionHeldReads: preAdmissionReads.size, preAdmissionHoldExpired }, lastStart, lastAdmission, responses: responses.slice(), |
| 66 | 75 | slot: slot === null ? null : { |
| 67 | 76 | ordinal: ordinal(slot), |
| @@ -123,6 +132,7 @@ const control = async (request: Request) => { | ||
| 123 | 132 | if (request.headers.get('x-fixture-control') !== secret) return new Response(null, { status: 404 }) |
| 124 | 133 | const action = new URL(request.url).pathname.slice('/__fixture/'.length) |
| 125 | 134 | if (action === 'status') return Response.json({ held, streamStarts, startRequests, startResponseReady: !!releaseStart, workspaceHeld, preAdmissionRenewed, preAdmissionHeldReads: preAdmissionReads.size, preAdmissionHoldExpired, admissionRequests }) |
| 135 | if (action === 'worker-callback-timings') return Response.json(callbackObserver.snapshot()) | |
| 126 | 136 | if (action === 'diagnostics') { |
| 127 | 137 | try { |
| 128 | 138 | const text = `${JSON.stringify(diagnosticSnapshot())}\n` |
| @@ -212,6 +222,9 @@ route.use(async (request: Request, next: () => Promise<Response | null> | Respon | ||
| 212 | 222 | } |
| 213 | 223 | const response = await next() |
| 214 | 224 | status = response?.status ?? null |
| 225 | // The genuine core Worker is already constructed. No constructor or frame is replaced. | |
| 226 | // Observation precedes the accepted response and the later native consumer attachment. | |
| 227 | if (startOrdinal !== null && status === 202) observeAcceptedWorker(startOrdinal) | |
| 215 | 228 | if (delayStart && request.method === 'POST' && pathname.endsWith('/workspace/execution') && response?.ok) { delayStart = false; await new Promise<void>(resolve => { releaseStart = resolve }) } |
| 216 | 229 | const executionEvents = /\/workspace\/execution\/[a-f0-9-]+\/events$/.test(pathname), workspaceEvents = pathname.endsWith('/workspace/events') |
| 217 | 230 | if (request.method !== 'GET' || !(executionEvents || workspaceEvents) || !response?.ok || !response.body) return response |
| @@ -247,7 +260,7 @@ route.use(async (request: Request, next: () => Promise<Response | null> | Respon | ||
| 247 | 260 | }) |
| 248 | 261 | const page = () => new Response(html, { headers: { 'content-type': 'text/html', 'cache-control': 'no-store' } }) |
| 249 | 262 | route.get('/incidents', page) |
| 250 | for (const action of ['status', 'diagnostics', 'release-start', 'arm-before-return', 'before-return', 'release-and-renew', 'before-admission', 'hold-completion-read-renewal', 'arm-pre-admission-renewal', 'renew-pre-admission-held-read', 'verify-admitted', 'release-workspace', 'read-only', 'revoke']) route.get(`/__fixture/${action}`, control) | |
| 263 | for (const action of ['status', 'diagnostics', 'worker-callback-timings', 'release-start', 'arm-before-return', 'before-return', 'release-and-renew', 'before-admission', 'hold-completion-read-renewal', 'arm-pre-admission-renewal', 'renew-pre-admission-held-read', 'verify-admitted', 'release-workspace', 'read-only', 'revoke']) route.get(`/__fixture/${action}`, control) | |
| 251 | 264 | const server = await route.serve({ hostname: '127.0.0.1', port: 0, nativeRoutes: false }) |
| 252 | 265 | if (process.env.EXECUTION_BROWSER_EVIDENCE) { await mkdir(process.env.EXECUTION_BROWSER_EVIDENCE, { recursive: true }); await writeFile(join(process.env.EXECUTION_BROWSER_EVIDENCE, 'compiled.html'), html) } |
| 253 | 266 | await writeFile(join(root, 'execution-credentials.json'), JSON.stringify({ url: `http://127.0.0.1:${server.port}`, key: seeded.key, sourceDigest: seeded.source.digest, token: fixture.senderToken, control: secret }), { mode: 0o600 }) |
| @@ -0,0 +1,5 @@ | ||
| 1 | self.onmessage = ({ data }) => { | |
| 2 | if (data !== 'emit') return | |
| 3 | self.onmessage = null | |
| 4 | for (const kind of ['event', 'result', 'finished', 'unrecognized']) self.postMessage({ kind, payload: 'private-callback-payload', bytes: new Uint8Array([1, 2, 3]) }) | |
| 5 | } | |
| @@ -214,6 +214,28 @@ try { | ||
| 214 | 214 | await control('revoke') |
| 215 | 215 | await first.wait("document.querySelector('[data-workspace-status]')?.textContent.trim() === 'refused'", 'Revocation must clear private execution content', 16000) |
| 216 | 216 | assert.equal(await first.evaluate("!!document.querySelector('[data-execution-id]') || document.body.textContent.includes(" + JSON.stringify(digest) + ")"), false) |
| 217 | const callbackTimings = await control('worker-callback-timings') | |
| 218 | assert.equal(callbackTimings.overflow, false) | |
| 219 | assert.equal(callbackTimings.unavailable, false) | |
| 220 | assert.equal(callbackTimings.observations.length, 2, 'Both genuine core Workers must have bounded callback observations') | |
| 221 | const nominalCallbacks = callbackTimings.observations[1] | |
| 222 | assert.equal(nominalCallbacks.ordinal, 2) | |
| 223 | assert.equal(nominalCallbacks.inCallback, false) | |
| 224 | assert.equal(nominalCallbacks.capped, false) | |
| 225 | assert(nominalCallbacks.costs.event.calls > 0) | |
| 226 | assert.equal(nominalCallbacks.costs.result.calls, 1) | |
| 227 | assert.equal(nominalCallbacks.costs.finished.calls, 1) | |
| 228 | for (const observation of callbackTimings.observations) { | |
| 229 | assert.equal(observation.inCallback, false) | |
| 230 | assert.equal(observation.capped, false) | |
| 231 | for (const kind of ['event', 'result', 'finished', 'other']) { | |
| 232 | const bucket = observation.costs[kind] | |
| 233 | assert(Number.isSafeInteger(bucket.calls) && bucket.calls >= 0 && bucket.calls <= 1000000) | |
| 234 | assert(Number.isFinite(bucket.totalMs) && bucket.totalMs >= 0 && bucket.totalMs <= Number.MAX_SAFE_INTEGER) | |
| 235 | assert(Number.isFinite(bucket.maxMs) && bucket.maxMs >= 0 && bucket.maxMs <= bucket.totalMs) | |
| 236 | } | |
| 237 | } | |
| 238 | await writeFile(join(evidence, 'worker-callback-timings.json'), `${JSON.stringify(callbackTimings)}\n`, { mode: 0o600, flag: 'wx' }) | |
| 217 | 239 | console.log('PASS actual execution browser: native live-before-return/GET renewal/stop/completion/explicit admission/read-only/revocation') |
| 218 | 240 | } |
| 219 | 241 | catch (error) { |
| @@ -0,0 +1,63 @@ | ||
| 1 | type CallbackKind = 'event' | 'result' | 'finished' | 'other' | |
| 2 | interface CallbackCost { calls: number, totalMs: number, maxMs: number } | |
| 3 | interface CallbackObservation { ordinal: number, inCallback: boolean, capped: boolean, costs: Record<CallbackKind, CallbackCost> } | |
| 4 | type CallbackMarker = { evidence: 'native-worker-callback', ordinal: number, kind: 'result' | 'finished', phase: 'entered', sinceObservationMs: number | null } | |
| 5 | | { evidence: 'native-worker-callback', ordinal: number, kind: 'result' | 'finished', phase: 'settled', callbackMs: number | null, sinceObservationMs: number | null } | |
| 6 | const scalar = (value: number) => Number.isFinite(value) && Math.abs(value) <= Number.MAX_SAFE_INTEGER ? value : null | |
| 7 | const field = (value: unknown, name: string): unknown => value && typeof value === 'object' ? Object.getOwnPropertyDescriptor(value, name)?.value : undefined | |
| 8 | const choice = <T extends string>(value: unknown, values: readonly T[]): T | null => values.find(item => item === value) ?? null | |
| 9 | ||
| 10 | /** Private fixture observation only. Measures synchronous callback work, never async completion. */ | |
| 11 | export function createExecutionCallbackObserver(marker: (value: CallbackMarker) => void, now: () => number = () => performance.now()) { | |
| 12 | const callbackObservations: CallbackObservation[] = [], observedWorkers = new WeakSet<Worker>() | |
| 13 | let callbackObservationOverflow = false, callbackObservationUnavailable = false | |
| 14 | const snapshot = () => ({ overflow: callbackObservationOverflow, unavailable: callbackObservationUnavailable, observations: structuredClone(callbackObservations) }) | |
| 15 | function observe(worker: Worker, acceptedOrdinal: number) { | |
| 16 | try { | |
| 17 | if (observedWorkers.has(worker)) return | |
| 18 | if (callbackObservations.length >= 2) { callbackObservationOverflow = true; return } | |
| 19 | const callback = worker.onmessage | |
| 20 | if (typeof callback !== 'function') { callbackObservationUnavailable = true; return } | |
| 21 | const cost = (): CallbackCost => ({ calls: 0, totalMs: 0, maxMs: 0 }) | |
| 22 | const observation: CallbackObservation = { ordinal: acceptedOrdinal, inCallback: false, capped: false, costs: { event: cost(), result: cost(), finished: cost(), other: cost() } } | |
| 23 | const installedAt = now() | |
| 24 | let markerCount = 0 | |
| 25 | const emitMarker = (value: CallbackMarker) => { | |
| 26 | if (markerCount >= 4) { observation.capped = true; return } | |
| 27 | markerCount++ | |
| 28 | marker(value) | |
| 29 | } | |
| 30 | worker.onmessage = function (event) { | |
| 31 | let began: number | null = null, kind: CallbackKind = 'other' | |
| 32 | try { | |
| 33 | kind = choice(field(event.data, 'kind'), ['event', 'result', 'finished']) ?? 'other' | |
| 34 | if (kind === 'result' || kind === 'finished') emitMarker({ evidence: 'native-worker-callback', ordinal: acceptedOrdinal, kind, phase: 'entered', sinceObservationMs: scalar(now() - installedAt) }) | |
| 35 | } | |
| 36 | catch { callbackObservationUnavailable = true } | |
| 37 | try { observation.inCallback = true; began = now() } | |
| 38 | catch { callbackObservationUnavailable = true } | |
| 39 | try { return callback.call(this, event) } | |
| 40 | finally { | |
| 41 | // Timing is observational: preserve the exact native callback outcome, | |
| 42 | // including an exception, even if collection or its closed console marker fails. | |
| 43 | try { | |
| 44 | const ended = now() | |
| 45 | observation.inCallback = false | |
| 46 | const elapsed = began === null ? null : scalar(ended - began) | |
| 47 | if (elapsed === null || elapsed < 0) callbackObservationUnavailable = true | |
| 48 | else { | |
| 49 | const bucket = observation.costs[kind] | |
| 50 | if (bucket.calls >= 1000000 || bucket.totalMs + elapsed > Number.MAX_SAFE_INTEGER) observation.capped = true | |
| 51 | else { bucket.calls++; bucket.totalMs += elapsed; bucket.maxMs = Math.max(bucket.maxMs, elapsed) } | |
| 52 | } | |
| 53 | if (kind === 'result' || kind === 'finished') emitMarker({ evidence: 'native-worker-callback', ordinal: acceptedOrdinal, kind, phase: 'settled', callbackMs: elapsed, sinceObservationMs: scalar(now() - installedAt) }) | |
| 54 | } | |
| 55 | catch { observation.inCallback = false; callbackObservationUnavailable = true } | |
| 56 | } | |
| 57 | } | |
| 58 | observedWorkers.add(worker); callbackObservations.push(observation) | |
| 59 | } | |
| 60 | catch { callbackObservationUnavailable = true } | |
| 61 | } | |
| 62 | return { observe, snapshot } | |
| 63 | } | |
| @@ -0,0 +1,127 @@ | ||
| 1 | import { expect, test } from 'bun:test' | |
| 2 | import { createExecutionCallbackObserver } from '../helpers/execution-worker-observation' | |
| 3 | ||
| 4 | const worker = () => new Worker(new URL('../fixtures/public-safety-access/execution-observer-worker.ts', import.meta.url), { env: {}, preload: [] }) | |
| 5 | ||
| 6 | test('real Worker frames and receiver survive private callback observation without persisting payloads', async () => { | |
| 7 | const source = worker(), markers: unknown[] = [] | |
| 8 | const observer = createExecutionCallbackObserver(value => markers.push(value)) | |
| 9 | const received: string[] = [] | |
| 10 | let timer: ReturnType<typeof setTimeout> | undefined | |
| 11 | try { | |
| 12 | const delivered = new Promise<void>((resolve, reject) => { | |
| 13 | timer = setTimeout(() => reject(new Error('Owned callback fixture did not deliver its frames')), 2000) | |
| 14 | source.onmessage = function (event) { | |
| 15 | try { | |
| 16 | expect(this).toBe(source) | |
| 17 | const frame = event.data as { kind: string, payload: string, bytes: Uint8Array } | |
| 18 | expect(frame.payload).toBe('private-callback-payload') | |
| 19 | expect(frame.bytes).toEqual(new Uint8Array([1, 2, 3])) | |
| 20 | received.push(frame.kind) | |
| 21 | if (received.length === 4) resolve() | |
| 22 | } | |
| 23 | catch (error) { reject(error) } | |
| 24 | } | |
| 25 | source.onerror = () => reject(new Error('Owned callback fixture Worker failed')) | |
| 26 | }) | |
| 27 | observer.observe(source, 1) | |
| 28 | source.postMessage('emit') | |
| 29 | await delivered | |
| 30 | expect(received).toEqual(['event', 'result', 'finished', 'unrecognized']) | |
| 31 | const snapshot = observer.snapshot() | |
| 32 | expect(snapshot.overflow).toBe(false) | |
| 33 | expect(snapshot.unavailable).toBe(false) | |
| 34 | expect(snapshot.observations).toHaveLength(1) | |
| 35 | expect(snapshot.observations[0]!.inCallback).toBe(false) | |
| 36 | expect(snapshot.observations[0]!.capped).toBe(false) | |
| 37 | for (const bucket of Object.values(snapshot.observations[0]!.costs)) { | |
| 38 | expect(bucket.calls).toBe(1) | |
| 39 | expect(Number.isFinite(bucket.totalMs) && bucket.totalMs >= 0).toBe(true) | |
| 40 | expect(bucket.maxMs).toBe(bucket.totalMs) | |
| 41 | } | |
| 42 | expect(markers).toHaveLength(4) | |
| 43 | expect(JSON.stringify({ snapshot, markers })).not.toContain('private-callback-payload') | |
| 44 | expect(JSON.stringify({ snapshot, markers })).not.toContain('bytes') | |
| 45 | } | |
| 46 | finally { clearTimeout(timer); source.terminate() } | |
| 47 | }) | |
| 48 | ||
| 49 | test('observer failures preserve the original callback return, throw identity and opaque fields', () => { | |
| 50 | const first = worker(), second = worker() | |
| 51 | const observer = createExecutionCallbackObserver(() => { throw new Error('Diagnostic marker unavailable') }) | |
| 52 | let getterCalls = 0, received: unknown | |
| 53 | const payload = { kind: 'result', get privateField() { getterCalls++; return 'unread-private-value' } } | |
| 54 | const event = new MessageEvent('message', { data: payload }), result = { originalReturn: true }, originalError = new Error('Original callback failure') | |
| 55 | try { | |
| 56 | first.onmessage = function (value) { expect(this).toBe(first); received = value.data; return result } | |
| 57 | observer.observe(first, 1) | |
| 58 | expect(first.onmessage!.call(first, event)).toBe(result) | |
| 59 | expect(received).toBe(payload) | |
| 60 | expect(getterCalls).toBe(0) | |
| 61 | second.onmessage = function (value) { expect(this).toBe(second); expect(value).toBe(event); throw originalError } | |
| 62 | observer.observe(second, 2) | |
| 63 | let thrown: unknown | |
| 64 | try { second.onmessage!.call(second, event) } | |
| 65 | catch (error) { thrown = error } | |
| 66 | expect(thrown).toBe(originalError) | |
| 67 | expect(observer.snapshot().unavailable).toBe(true) | |
| 68 | expect(observer.snapshot().observations.map(value => value.costs.result.calls)).toEqual([1, 1]) | |
| 69 | expect(JSON.stringify(observer.snapshot())).not.toContain('unread-private-value') | |
| 70 | expect(getterCalls).toBe(0) | |
| 71 | } | |
| 72 | finally { first.terminate(); second.terminate() } | |
| 73 | }) | |
| 74 | ||
| 75 | test('the observer caps actual Worker reservations without changing later callbacks and returns detached snapshots', () => { | |
| 76 | const first = worker(), second = worker(), third = worker() | |
| 77 | const observer = createExecutionCallbackObserver(() => {}) | |
| 78 | const callback = () => {} | |
| 79 | try { | |
| 80 | first.onmessage = callback; second.onmessage = callback; third.onmessage = callback | |
| 81 | observer.observe(first, 1); observer.observe(first, 99); observer.observe(second, 2); observer.observe(third, 3) | |
| 82 | const snapshot = observer.snapshot() | |
| 83 | expect(snapshot.overflow).toBe(true) | |
| 84 | expect(snapshot.unavailable).toBe(false) | |
| 85 | expect(snapshot.observations.map(value => value.ordinal)).toEqual([1, 2]) | |
| 86 | expect(third.onmessage).toBe(callback) | |
| 87 | snapshot.observations[0]!.costs.event.calls = 9999 | |
| 88 | expect(observer.snapshot().observations[0]!.costs.event.calls).toBe(0) | |
| 89 | } | |
| 90 | finally { first.terminate(); second.terminate(); third.terminate() } | |
| 91 | }) | |
| 92 | ||
| 93 | test('marker latency and diagnostic exceptions cannot inflate the original synchronous callback duration', () => { | |
| 94 | const source = worker(), returned = { original: true }, settled: number[] = [] | |
| 95 | let clock = 0 | |
| 96 | const observer = createExecutionCallbackObserver(value => { | |
| 97 | if (value.phase === 'settled' && value.callbackMs !== null) settled.push(value.callbackMs) | |
| 98 | clock += 1000 | |
| 99 | throw new Error('Slow diagnostic sink failed') | |
| 100 | }, () => clock) | |
| 101 | try { | |
| 102 | source.onmessage = () => { clock += 7; return returned } | |
| 103 | observer.observe(source, 1) | |
| 104 | expect(source.onmessage!.call(source, new MessageEvent('message', { data: { kind: 'result' } }))).toBe(returned) | |
| 105 | expect(observer.snapshot().observations[0]!.costs.result).toEqual({ calls: 1, totalMs: 7, maxMs: 7 }) | |
| 106 | expect(settled).toEqual([7]) | |
| 107 | expect(observer.snapshot().unavailable).toBe(true) | |
| 108 | } | |
| 109 | finally { source.terminate() } | |
| 110 | }) | |
| 111 | ||
| 112 | test('repeated terminal-kind messages cannot produce an unbounded diagnostic console stream', () => { | |
| 113 | const source = worker(), markers: unknown[] = [] | |
| 114 | const observer = createExecutionCallbackObserver(value => markers.push(value)) | |
| 115 | let originalCalls = 0 | |
| 116 | try { | |
| 117 | source.onmessage = () => { originalCalls++ } | |
| 118 | observer.observe(source, 1) | |
| 119 | for (let index = 0; index < 10; index++) source.onmessage!.call(source, new MessageEvent('message', { data: { kind: 'result' } })) | |
| 120 | expect(originalCalls).toBe(10) | |
| 121 | expect(markers).toHaveLength(4) | |
| 122 | expect(observer.snapshot().observations[0]!.capped).toBe(true) | |
| 123 | expect(observer.snapshot().observations[0]!.costs.result.calls).toBe(10) | |
| 124 | expect(observer.snapshot().unavailable).toBe(false) | |
| 125 | } | |
| 126 | finally { source.terminate() } | |
| 127 | }) | |