also looking at this
test: retain execution HTTP lifecycle evidence
#95
3 files
+150
-58
| @@ -2,21 +2,36 @@ import assert from 'node:assert/strict' | ||
| 2 | 2 | import { writeFile } from 'node:fs/promises' |
| 3 | 3 | import { join, resolve } from 'node:path' |
| 4 | 4 | import type { ExecutionView } from '../../../resources/functions/drone/public-safety/execution/types' |
| 5 | import { setupHandoverFixture } from './handover-setup' | |
| 6 | 5 | |
| 7 | 6 | const scenario = Bun.argv[2] |
| 8 | 7 | assert(['limits', 'preflight', 'middleware', 'chunk', 'admission'].includes(scenario!)) |
| 9 | const f = await setupHandoverFixture() | |
| 8 | const began = performance.now() | |
| 9 | let observations = 0 | |
| 10 | async function observe<T>(phase: string, work: () => T | Promise<T>): Promise<T> { | |
| 11 | const emit = (outcome: 'began' | 'settled' | 'rejected') => { | |
| 12 | if (observations >= 128) return | |
| 13 | // Diagnostic output must not replace an actual native failure. | |
| 14 | try { console.log(JSON.stringify({ oracle: 'execution-http-phase', scenario, pid: process.pid, ordinal: observations++, phase, outcome, elapsedMs: performance.now() - began })) } | |
| 15 | catch {} | |
| 16 | } | |
| 17 | emit('began') | |
| 18 | try { const result = await work(); emit('settled'); return result } | |
| 19 | catch (error) { emit('rejected'); throw error } | |
| 20 | } | |
| 21 | try { console.log(JSON.stringify({ oracle: 'execution-http-runtime', scenario, pid: process.pid, bunVersion: Bun.version, executable: process.execPath, platform: process.platform, arch: process.arch })) } | |
| 22 | catch {} | |
| 23 | const { setupHandoverFixture } = await observe('setup-module', () => import('./handover-setup')) | |
| 24 | const f = await observe('setup', setupHandoverFixture) | |
| 10 | 25 | const source = resolve(import.meta.dir, '../../..') |
| 11 | await writeFile(join(process.cwd(), 'app/Gates.ts'), `export { default } from ${JSON.stringify(join(source, 'app/Gates.ts'))}`) | |
| 12 | const { initializeAuthorization } = await import('@stacksjs/auth/policy') | |
| 13 | await initializeAuthorization() | |
| 14 | const { Gate } = await import('@stacksjs/auth/gate') | |
| 15 | const { getCurrentRequest, route } = await import('@stacksjs/router') | |
| 16 | const { emit, setReplayBuffer, stopServer } = await import('@stacksjs/realtime') | |
| 17 | const { executionOwner } = await import('../../../resources/functions/drone/public-safety/execution/state') | |
| 18 | const { executionWords } = await import('../../../resources/functions/drone/public-safety/execution/protocol') | |
| 19 | await route.register(join(source, 'routes/api.ts'), { prefix: '/api' }) | |
| 26 | await observe('gate-override', () => writeFile(join(process.cwd(), 'app/Gates.ts'), `export { default } from ${JSON.stringify(join(source, 'app/Gates.ts'))}`)) | |
| 27 | const { initializeAuthorization } = await observe('authorization-module', () => import('@stacksjs/auth/policy')) | |
| 28 | await observe('authorization', initializeAuthorization) | |
| 29 | const { Gate } = await observe('gate-module', () => import('@stacksjs/auth/gate')) | |
| 30 | const { getCurrentRequest, route } = await observe('router-module', () => import('@stacksjs/router')) | |
| 31 | const { emit, setReplayBuffer, stopServer } = await observe('realtime-module', () => import('@stacksjs/realtime')) | |
| 32 | const { executionOwner } = await observe('execution-state-module', () => import('../../../resources/functions/drone/public-safety/execution/state')) | |
| 33 | const { executionWords } = await observe('protocol-module', () => import('../../../resources/functions/drone/public-safety/execution/protocol')) | |
| 34 | await observe('route-register', () => route.register(join(source, 'routes/api.ts'), { prefix: '/api' })) | |
| 20 | 35 | let deadline = 0, crossed = 0 |
| 21 | 36 | const cross = async () => { |
| 22 | 37 | assert(Date.now() < deadline, 'Actual native response/check must be current before the wall crossing') |
| @@ -30,7 +45,7 @@ route.use(async (request: Request, next: () => Promise<Response | null> | Respon | ||
| 30 | 45 | return response |
| 31 | 46 | }) |
| 32 | 47 | const nativeRoutes = process.env.INCIDENT_NATIVE_ROUTES === 'true' |
| 33 | const server = await route.serve({ hostname: '127.0.0.1', port: 0, nativeRoutes }) | |
| 48 | const server = await observe('server-start', () => route.serve({ hostname: '127.0.0.1', port: 0, nativeRoutes })) | |
| 34 | 49 | type DiagnosticSlot = { control: Int32Array, subscriber: unknown, worker: Worker | null, producer: string, admission: string, phase: string, reason: string | null } |
| 35 | 50 | let slot: DiagnosticSlot | undefined, started: ExecutionView | undefined |
| 36 | 51 | const readers = new Set<ReadableStreamDefaultReader<Uint8Array>>() |
| @@ -51,13 +66,13 @@ try { | ||
| 51 | 66 | // Chunk/admission must first complete an actual run. Give that setup margin; |
| 52 | 67 | // the subsequent native release still has to cross this genuine wall deadline. |
| 53 | 68 | const retentionMs = scenario === 'chunk' || scenario === 'admission' ? 20000 : expiring ? 8000 : 86400000 |
| 54 | const seeded = await f.seed(`execution-http-${scenario}`, retentionMs) | |
| 69 | const seeded = await observe('seed', () => f.seed(`execution-http-${scenario}`, retentionMs)) | |
| 55 | 70 | deadline = Date.parse(seeded.config.retention.expiresAt) |
| 56 | 71 | const base = `http://127.0.0.1:${server.port}/api/public-safety/incidents/${seeded.key}/workspace` |
| 57 | 72 | const headers = { authorization: `Bearer ${f.senderToken}` } |
| 58 | const workspace = await fetch(base, { headers }); assert.equal(workspace.status, 200) | |
| 59 | const { scope } = await workspace.json() as { scope: string } | |
| 60 | const initialVersion = Number((await f.snapshot(seeded.key)).incident!.version) | |
| 73 | const workspace = await observe('workspace-request', () => fetch(base, { headers })); assert.equal(workspace.status, 200) | |
| 74 | const { scope } = await observe('workspace-body', () => workspace.json()) as { scope: string } | |
| 75 | const initialVersion = Number((await observe('initial-snapshot', () => f.snapshot(seeded.key))).incident!.version) | |
| 61 | 76 | const statusUrl = `${base}/execution?scope=${scope}` |
| 62 | 77 | const send = (suffix: string, body: unknown, extra: Record<string, string> = {}) => fetch(`${base}/execution${suffix}`, { method: 'POST', headers: { ...headers, 'content-type': 'application/json', ...extra }, body: JSON.stringify(body) }) |
| 63 | 78 | cleanup = async () => { |
| @@ -74,35 +89,35 @@ try { | ||
| 74 | 89 | } |
| 75 | 90 | if (scenario === 'preflight') { |
| 76 | 91 | stall = true |
| 77 | const response = await fetch(statusUrl, { headers: { ...headers, 'x-fixture-execution-stall': '1' } }) | |
| 92 | const response = await observe('preflight-request', () => fetch(statusUrl, { headers: { ...headers, 'x-fixture-execution-stall': '1' } })) | |
| 78 | 93 | assert.equal(response.status, 404) |
| 79 | 94 | assert(held) |
| 80 | 95 | assert.equal(executionOwner().pendingChecks, 1, 'Timed-out actual HTTP preflight remains charged until genuine native settlement') |
| 81 | 96 | resume!(); stall = false |
| 82 | await wait(() => executionOwner().pendingChecks === 0, 'Native preflight must settle before its shared budget is released') | |
| 97 | await observe('preflight-settlement', () => wait(() => executionOwner().pendingChecks === 0, 'Native preflight must settle before its shared budget is released')) | |
| 83 | 98 | assert.equal((await fetch(statusUrl, { headers })).status, 200) |
| 84 | 99 | assert.equal(executionOwner().slot, null) |
| 85 | 100 | } |
| 86 | 101 | else if (scenario === 'middleware') { |
| 87 | const response = await fetch(statusUrl, { headers: { ...headers, 'x-fixture-execution-middleware': '1' } }) | |
| 102 | const response = await observe('middleware-request', () => fetch(statusUrl, { headers: { ...headers, 'x-fixture-execution-middleware': '1' } })) | |
| 88 | 103 | assert.equal(response.status, 404, 'Null status response retains preflight deadline across awaited native middleware') |
| 89 | 104 | assert.equal(crossed, 1) |
| 90 | 105 | assert.deepEqual(Object.keys(await response.json() as object), ['error']) |
| 91 | 106 | } |
| 92 | 107 | else { |
| 93 | const response = await send('', { intent: crypto.randomUUID(), scope, scenario: 'nominal', seconds: 8 }) | |
| 108 | const response = await observe('execution-start', () => send('', { intent: crypto.randomUUID(), scope, scenario: 'nominal', seconds: 8 })) | |
| 94 | 109 | assert.equal(response.status, 202); started = await response.json() as ExecutionView |
| 95 | 110 | slot = executionOwner().slot as DiagnosticSlot |
| 96 | 111 | const events = `${base}/execution/${started.execution}/events?scope=${scope}` |
| 97 | 112 | const channel = `private-public-safety.execution.${started.execution}` |
| 98 | const open = async (extra: Record<string, string> = {}) => { | |
| 113 | const open = (extra: Record<string, string> = {}) => observe('events-open', async () => { | |
| 99 | 114 | const abort = new AbortController(); aborts.add(abort) |
| 100 | 115 | const result = await route.handleRequest(new Request(events, { headers: { ...headers, ...extra }, signal: abort.signal }) as Parameters<typeof route.handleRequest>[0]) |
| 101 | 116 | assert.equal(result?.status, 200) |
| 102 | 117 | const reader = result!.body!.getReader(); readers.add(reader) |
| 103 | 118 | assert(new TextDecoder().decode((await reader.read()).value).includes(': connected')) |
| 104 | 119 | return { reader, abort } |
| 105 | } | |
| 120 | }) | |
| 106 | 121 | const drainClosed = async (reader: ReadableStreamDefaultReader<Uint8Array>) => { |
| 107 | 122 | for (let count = 0; count < 100; count++) if ((await reader.read()).done) return |
| 108 | 123 | assert.fail('Native stream must close without unbounded pending frames') |
| @@ -141,26 +156,30 @@ try { | ||
| 141 | 156 | } |
| 142 | 157 | else { |
| 143 | 158 | const live = await open({ 'x-fixture-execution-chunk': '1' }) |
| 144 | let complete: ExecutionView | undefined, text = '' | |
| 145 | const decoder = new TextDecoder() | |
| 146 | while (!complete) { | |
| 147 | const part = await live.reader.read() | |
| 148 | if (part.done) console.log(JSON.stringify({ oracle: 'terminal-preparation-ended', scenario, nativeRoutes, remainingMs: deadline - Date.now(), phase: slot.phase, reason: slot.reason, producer: slot.producer, credit: Atomics.load(slot.control, executionWords.credit), emitted: Atomics.load(slot.control, executionWords.emitted), returned: Atomics.load(slot.control, executionWords.returned), pending: executionOwner().pendingChecks })) | |
| 149 | assert(!part.done) | |
| 150 | text += decoder.decode(part.value, { stream: true }) | |
| 151 | for (let end = text.indexOf('\n\n'); end >= 0; end = text.indexOf('\n\n')) { | |
| 152 | const frame = text.slice(0, end); text = text.slice(end + 2) | |
| 153 | if (!frame.startsWith('data: ')) continue | |
| 154 | const value = JSON.parse(frame.slice(6)) as { data: ExecutionView } | |
| 155 | if (value.data.phase === 'completed') complete = value.data | |
| 159 | const activeSlot = slot | |
| 160 | const complete = await observe('producer-terminal', async () => { | |
| 161 | let complete: ExecutionView | undefined, text = '' | |
| 162 | const decoder = new TextDecoder() | |
| 163 | while (!complete) { | |
| 164 | const part = await live.reader.read() | |
| 165 | if (part.done) console.log(JSON.stringify({ oracle: 'terminal-preparation-ended', scenario, nativeRoutes, remainingMs: deadline - Date.now(), phase: activeSlot.phase, reason: activeSlot.reason, producer: activeSlot.producer, credit: Atomics.load(activeSlot.control, executionWords.credit), emitted: Atomics.load(activeSlot.control, executionWords.emitted), returned: Atomics.load(activeSlot.control, executionWords.returned), pending: executionOwner().pendingChecks })) | |
| 166 | assert(!part.done) | |
| 167 | text += decoder.decode(part.value, { stream: true }) | |
| 168 | for (let end = text.indexOf('\n\n'); end >= 0; end = text.indexOf('\n\n')) { | |
| 169 | const frame = text.slice(0, end); text = text.slice(end + 2) | |
| 170 | if (!frame.startsWith('data: ')) continue | |
| 171 | const value = JSON.parse(frame.slice(6)) as { data: ExecutionView } | |
| 172 | if (value.data.phase === 'completed') complete = value.data | |
| 173 | } | |
| 156 | 174 | } |
| 157 | } | |
| 175 | return complete | |
| 176 | }) | |
| 158 | 177 | console.log(JSON.stringify({ oracle: 'terminal-preparation-completed', scenario, nativeRoutes, remainingMs: deadline - Date.now(), phase: slot.phase, reason: slot.reason, producer: slot.producer, credit: Atomics.load(slot.control, executionWords.credit), emitted: Atomics.load(slot.control, executionWords.emitted), returned: Atomics.load(slot.control, executionWords.returned) })) |
| 159 | 178 | if (scenario === 'chunk') { |
| 160 | 179 | live.abort.abort(); await live.reader.cancel() |
| 161 | 180 | // Leave no native queue idle while waiting. Reconnect the SAME retained |
| 162 | 181 | // execution within one heartbeat of expiry, then hold its final read. |
| 163 | await Bun.sleep(Math.max(1, deadline - Date.now() - 1500)) | |
| 182 | await observe('terminal-reconnect-wait', () => Bun.sleep(Math.max(1, deadline - Date.now() - 1500))) | |
| 164 | 183 | const final = await open({ 'x-fixture-execution-final-chunk': '1' }) |
| 165 | 184 | await Bun.sleep(10) |
| 166 | 185 | await wait(() => executionOwner().pendingChecks === 0, 'Terminal reconnect setup native checks must settle', 500) |
| @@ -169,7 +188,7 @@ try { | ||
| 169 | 188 | f.database.db.primary.setTransactionDefaults({ afterCommit: async () => { |
| 170 | 189 | if (!crossed && getCurrentRequest()?.headers.has('x-fixture-execution-final-chunk')) await cross() |
| 171 | 190 | } }) |
| 172 | assert.equal((await final.reader.read()).done, true, 'Opaque chunk is withheld when its actual native check COMMIT crosses retention') | |
| 191 | assert.equal((await observe('final-chunk-read', () => final.reader.read())).done, true, 'Opaque chunk is withheld when its actual native check COMMIT crosses retention') | |
| 173 | 192 | assert.equal(crossed, 1) |
| 174 | 193 | } |
| 175 | 194 | else { |
| @@ -182,7 +201,7 @@ try { | ||
| 182 | 201 | f.database.db.primary.setTransactionDefaults({ afterCommit: async () => { |
| 183 | 202 | if (inserted && !crossed && getCurrentRequest()?.headers.has('x-fixture-execution-admission')) await cross() |
| 184 | 203 | } }) |
| 185 | const result = await send('/admission', { execution: started.execution, scope, digest: complete.completion!.digest, expectedVersion: initialVersion }, { 'x-fixture-execution-admission': '1' }) | |
| 204 | const result = await observe('admission-request', () => send('/admission', { execution: started!.execution, scope, digest: complete.completion!.digest, expectedVersion: initialVersion }, { 'x-fixture-execution-admission': '1' })) | |
| 186 | 205 | assert.equal(result.status, 404, 'A genuinely committed admission cannot release past its authority deadline') |
| 187 | 206 | assert.equal(crossed, 1) |
| 188 | 207 | assert(inserted, 'The actual checked writer inserted the source before its native COMMIT wait') |
| @@ -213,15 +232,21 @@ finally { | ||
| 213 | 232 | f.database.db.primary.setTransactionDefaults({ afterCommit: undefined }) |
| 214 | 233 | removeQueries?.() |
| 215 | 234 | for (const abort of aborts) abort.abort() |
| 216 | await Promise.allSettled([...readers].map(reader => reader.cancel())) | |
| 217 | await cleanup?.() | |
| 218 | if (slot?.producer === 'pending') { | |
| 219 | const until = Date.now() + 4000 | |
| 220 | while (slot.producer === 'pending' && Date.now() < until) await Bun.sleep(20) | |
| 221 | if (slot.producer === 'pending') slot.worker?.terminate() | |
| 222 | } | |
| 223 | const until = Date.now() + 5000 | |
| 224 | while (executionOwner().pendingChecks && Date.now() < until) await Bun.sleep(10) | |
| 235 | await observe('cleanup-readers', () => Promise.allSettled([...readers].map(reader => reader.cancel()))) | |
| 236 | await observe('cleanup-execution', async () => { await cleanup?.() }) | |
| 237 | await observe('cleanup-producer', async () => { | |
| 238 | if (slot?.producer === 'pending') { | |
| 239 | const until = Date.now() + 4000 | |
| 240 | while (slot.producer === 'pending' && Date.now() < until) await Bun.sleep(20) | |
| 241 | if (slot.producer === 'pending') slot.worker?.terminate() | |
| 242 | } | |
| 243 | }) | |
| 244 | await observe('cleanup-pending-checks', async () => { | |
| 245 | const until = Date.now() + 5000 | |
| 246 | while (executionOwner().pendingChecks && Date.now() < until) await Bun.sleep(10) | |
| 247 | }) | |
| 225 | 248 | assert.equal(executionOwner().pendingChecks, 0) |
| 226 | await server.stop(true); await stopServer(); await f.database.closeDatabaseConnection() | |
| 249 | await observe('cleanup-server', async () => { await server.stop(true) }) | |
| 250 | await observe('cleanup-realtime', stopServer) | |
| 251 | await observe('cleanup-database', () => f.database.closeDatabaseConnection()) | |
| 227 | 252 | } |