mirror of
https://github.com/stablyai/orca.git
synced 2026-11-04 02:57:36 +00:00
refactor(orcad): orcad artifacts live under orcad/ and node-runtime/ (#27963)
Co-authored-by: m4air <m4air@Mac.localdomain>
This commit is contained in:
OrcaWin
m4air
@@ -0,0 +1,22 @@
|
||||
import { lstat } from 'node:fs/promises'
|
||||
|
||||
/** Recover beside corrupt entries; a published path may still belong to another reader. */
|
||||
export async function findOrcadCachePath(
|
||||
candidate: (attempt: number) => string,
|
||||
isValid: (path: string) => Promise<boolean>
|
||||
): Promise<{ path: string; verified: boolean }> {
|
||||
for (let attempt = 0; ; attempt++) {
|
||||
const path = candidate(attempt)
|
||||
if (await isValid(path)) {
|
||||
return { path, verified: true }
|
||||
}
|
||||
try {
|
||||
await lstat(path)
|
||||
} catch (error) {
|
||||
if (error instanceof Error && 'code' in error && error.code === 'ENOENT') {
|
||||
return { path, verified: false }
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { mkdir, mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import { runProcess } from '@orca/process-host'
|
||||
import {
|
||||
NODE_RUNTIME_ASSETS,
|
||||
NODE_RUNTIME_PIN,
|
||||
nodeRuntimeExecutablePath
|
||||
} from '../../shared/node-runtime-pin'
|
||||
import { materializeNodeRuntimeArchive } from './pinned-runtime-materializer'
|
||||
|
||||
const TARGET = 'linux-x64-glibc' as const
|
||||
const originalAsset = { ...NODE_RUNTIME_ASSETS[TARGET] }
|
||||
let root = ''
|
||||
|
||||
function sha256(bytes: Uint8Array): string {
|
||||
return createHash('sha256').update(bytes).digest('hex')
|
||||
}
|
||||
|
||||
function fetcherFor(body: Uint8Array, signals: AbortSignal[] = []): typeof fetch {
|
||||
return vi.fn<typeof fetch>(async (_url, options) => {
|
||||
if (options?.signal) {
|
||||
signals.push(options.signal)
|
||||
}
|
||||
return new Response(Buffer.from(body), { status: 200 })
|
||||
})
|
||||
}
|
||||
|
||||
/** A real .tar.gz laid out like the official one, so extraction runs the host's tar. */
|
||||
async function nodeDistArchiveFixture(executable: Uint8Array): Promise<Uint8Array> {
|
||||
const staging = join(root, 'staging')
|
||||
const member = nodeRuntimeExecutablePath(TARGET, NODE_RUNTIME_ASSETS[TARGET].archive)
|
||||
await mkdir(join(staging, member, '..'), { recursive: true })
|
||||
await writeFile(join(staging, member), executable)
|
||||
await writeFile(join(staging, member.split('/')[0]!, 'README.md'), 'not extracted')
|
||||
const archivePath = join(root, 'archive.tar.gz')
|
||||
const result = await runProcess({
|
||||
program: 'tar',
|
||||
args: ['-czf', archivePath, '-C', staging, member.split('/')[0]!]
|
||||
})
|
||||
expect(result.code, result.stderr).toBe(0)
|
||||
return new Uint8Array(await readFile(archivePath))
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-node-runtime-materializer-'))
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
Object.assign(NODE_RUNTIME_ASSETS[TARGET], originalAsset)
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe.skipIf(process.platform === 'win32')('pinned Node runtime materializer', () => {
|
||||
it('keeps the verified archive for upload and refetches only when it is corrupted', async () => {
|
||||
const archive = await nodeDistArchiveFixture(new TextEncoder().encode('node'))
|
||||
Object.assign(NODE_RUNTIME_ASSETS[TARGET], { archiveSha256: sha256(archive) })
|
||||
const cacheRoot = join(root, 'cache')
|
||||
const signals: AbortSignal[] = []
|
||||
const fetcher = fetcherFor(archive, signals)
|
||||
|
||||
const cached = await materializeNodeRuntimeArchive(TARGET, cacheRoot, { fetcher })
|
||||
expect(cached.endsWith(originalAsset.archive)).toBe(true)
|
||||
expect(fetcher).toHaveBeenCalledWith(
|
||||
expect.stringContaining(`/v${NODE_RUNTIME_PIN.version}/${originalAsset.archive}`),
|
||||
expect.objectContaining({ redirect: 'follow' })
|
||||
)
|
||||
expect(await materializeNodeRuntimeArchive(TARGET, cacheRoot, { fetcher })).toBe(cached)
|
||||
expect(fetcher).toHaveBeenCalledOnce()
|
||||
|
||||
await writeFile(cached, 'torn')
|
||||
const repaired = await materializeNodeRuntimeArchive(TARGET, cacheRoot, { fetcher })
|
||||
expect(repaired).not.toBe(cached)
|
||||
expect(sha256(new Uint8Array(await readFile(repaired)))).toBe(sha256(archive))
|
||||
expect(signals).toHaveLength(2)
|
||||
expect(signals.every((signal) => !signal.aborted)).toBe(true)
|
||||
})
|
||||
|
||||
it('refuses an archive that does not match the pin before caching it', async () => {
|
||||
const archive = await nodeDistArchiveFixture(new TextEncoder().encode('node'))
|
||||
await expect(
|
||||
materializeNodeRuntimeArchive(TARGET, join(root, 'cache'), { fetcher: fetcherFor(archive) })
|
||||
).rejects.toThrow('Node archive checksum mismatch')
|
||||
})
|
||||
})
|
||||
|
||||
describe('pinned runtime archive request ownership', () => {
|
||||
it.each(['http-error', 'oversized-declaration', 'staging-collision'] as const)(
|
||||
'retires the owned %s request without cancelling its caller',
|
||||
async (failure) => {
|
||||
const cacheRoot = join(root, 'cache')
|
||||
const archiveRoot = join(cacheRoot, 'node', 'archives')
|
||||
const parent = new AbortController()
|
||||
const signals: AbortSignal[] = []
|
||||
const cancel = vi.fn()
|
||||
const fetcher = vi.fn<typeof fetch>(async (_url, options) => {
|
||||
if (options?.signal) {
|
||||
signals.push(options.signal)
|
||||
}
|
||||
if (failure === 'staging-collision') {
|
||||
const stage = (await readdir(archiveRoot)).find((name) => name.startsWith('.download-'))
|
||||
if (!stage) {
|
||||
throw new Error('Missing download staging directory')
|
||||
}
|
||||
await mkdir(join(archiveRoot, stage, originalAsset.archive))
|
||||
}
|
||||
return new Response(new ReadableStream({ cancel }), {
|
||||
status: failure === 'http-error' ? 404 : 200,
|
||||
statusText: 'Unavailable',
|
||||
headers:
|
||||
failure === 'oversized-declaration'
|
||||
? { 'content-length': String(200 * 1024 * 1024 + 1) }
|
||||
: {}
|
||||
})
|
||||
})
|
||||
const pending = materializeNodeRuntimeArchive(TARGET, cacheRoot, {
|
||||
fetcher,
|
||||
signal: parent.signal
|
||||
})
|
||||
await (failure === 'staging-collision'
|
||||
? expect(pending).rejects.toMatchObject({ code: 'EEXIST' })
|
||||
: expect(pending).rejects.toThrow(
|
||||
failure === 'http-error'
|
||||
? 'Node download failed: 404 Unavailable'
|
||||
: 'Node download exceeded the archive size limit'
|
||||
))
|
||||
expect(cancel).toHaveBeenCalledOnce()
|
||||
expect(signals).toHaveLength(1)
|
||||
expect(signals[0]?.aborted).toBe(true)
|
||||
expect(parent.signal.aborted).toBe(false)
|
||||
expect(await readdir(archiveRoot)).toEqual([])
|
||||
}
|
||||
)
|
||||
|
||||
it('keeps the caller cancellation reason and removes the incomplete stage', async () => {
|
||||
const cacheRoot = join(root, 'cache')
|
||||
const parent = new AbortController()
|
||||
const reason = new Error('deployment cancelled')
|
||||
const cancel = vi.fn()
|
||||
const fetcher = vi.fn<typeof fetch>(async () => {
|
||||
queueMicrotask(() => parent.abort(reason))
|
||||
return new Response(new ReadableStream({ cancel }))
|
||||
})
|
||||
await expect(
|
||||
materializeNodeRuntimeArchive(TARGET, cacheRoot, { fetcher, signal: parent.signal })
|
||||
).rejects.toBe(reason)
|
||||
expect(parent.signal.reason).toBe(reason)
|
||||
expect(cancel).toHaveBeenCalledOnce()
|
||||
expect(await readdir(join(cacheRoot, 'node', 'archives'))).toEqual([])
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,299 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { access, mkdir, mkdtemp, readFile, readdir, rm, stat, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
NODE_RUNTIME_ASSETS,
|
||||
NODE_RUNTIME_PIN,
|
||||
nodeRuntimeExecutablePath,
|
||||
SERVER_TARGETS,
|
||||
type ServerTarget
|
||||
} from '../../shared/node-runtime-pin'
|
||||
import { setMainHttpClient } from '../network/http-client'
|
||||
import { runProcess } from '@orca/process-host'
|
||||
import { materializeCachedNodeRuntime } from './pinned-runtime-materializer'
|
||||
import { RUNTIME_ARCHIVE_RETRY_DELAYS_MS } from './runtime-archive-download'
|
||||
|
||||
const extraction = vi.hoisted(() => ({ executable: new Uint8Array(), member: '' }))
|
||||
|
||||
vi.mock('@orca/process-host', () => ({
|
||||
runProcess: vi.fn(async (spec: { args: string[] }) => {
|
||||
const flag = spec.args.includes('-C') ? '-C' : '-d'
|
||||
const extracted = join(spec.args[spec.args.indexOf(flag) + 1]!, ...extraction.member.split('/'))
|
||||
await mkdir(join(extracted, '..'), { recursive: true })
|
||||
await writeFile(extracted, extraction.executable)
|
||||
return { code: 0, stdout: '', stderr: '' }
|
||||
})
|
||||
}))
|
||||
|
||||
const TARGET = 'linux-x64-glibc' as const
|
||||
const originalAssets = structuredClone(NODE_RUNTIME_ASSETS)
|
||||
let cacheRoot = ''
|
||||
|
||||
function sha256(bytes: Uint8Array): string {
|
||||
return createHash('sha256').update(bytes).digest('hex')
|
||||
}
|
||||
|
||||
function runtimeDirFor(target: ServerTarget = TARGET): string {
|
||||
return join(cacheRoot, 'node', NODE_RUNTIME_ASSETS[target].executableSha256)
|
||||
}
|
||||
|
||||
function pin(target: ServerTarget, archive: Uint8Array, executable?: Uint8Array): void {
|
||||
NODE_RUNTIME_ASSETS[target].archiveSha256 = sha256(archive)
|
||||
if (executable) {
|
||||
NODE_RUNTIME_ASSETS[target].executableSha256 = sha256(executable)
|
||||
}
|
||||
extraction.member = nodeRuntimeExecutablePath(target, NODE_RUNTIME_ASSETS[target].archive)
|
||||
}
|
||||
|
||||
function responseFetcher(body: Uint8Array, declaredLength = body.byteLength): typeof fetch {
|
||||
return vi.fn<typeof fetch>(
|
||||
async () =>
|
||||
new Response(Buffer.from(body), {
|
||||
status: 200,
|
||||
headers: { 'content-length': String(declaredLength) }
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
extraction.member = nodeRuntimeExecutablePath(TARGET, NODE_RUNTIME_ASSETS[TARGET].archive)
|
||||
cacheRoot = await mkdtemp(join(tmpdir(), 'orca-node-runtime-executable-'))
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
vi.useRealTimers()
|
||||
vi.unstubAllEnvs()
|
||||
setMainHttpClient(null)
|
||||
for (const target of SERVER_TARGETS) {
|
||||
Object.assign(NODE_RUNTIME_ASSETS[target], originalAssets[target])
|
||||
}
|
||||
await rm(cacheRoot, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe('materializeCachedNodeRuntime', () => {
|
||||
it('caches a Windows executable under its upstream name, keyed by its digest', async () => {
|
||||
const target = 'win32-x64' as const
|
||||
const archive = new TextEncoder().encode('windows archive')
|
||||
const executable = new TextEncoder().encode('windows executable')
|
||||
extraction.executable = executable
|
||||
pin(target, archive, executable)
|
||||
const runtimePath = await materializeCachedNodeRuntime(target, cacheRoot, {
|
||||
fetcher: responseFetcher(archive)
|
||||
})
|
||||
expect(runtimePath).toBe(join(runtimeDirFor(target), 'node.exe'))
|
||||
expect(await readFile(runtimePath)).toEqual(Buffer.from(executable))
|
||||
})
|
||||
|
||||
it('reuses a checksum-valid cached runtime without fetching', async () => {
|
||||
const runtime = new TextEncoder().encode('cached node')
|
||||
NODE_RUNTIME_ASSETS[TARGET].executableSha256 = sha256(runtime)
|
||||
const runtimeDir = runtimeDirFor()
|
||||
const runtimePath = join(runtimeDir, 'node')
|
||||
await mkdir(runtimeDir, { recursive: true })
|
||||
await writeFile(runtimePath, runtime)
|
||||
const fetcher = vi.fn<typeof fetch>()
|
||||
|
||||
await expect(materializeCachedNodeRuntime(TARGET, cacheRoot, { fetcher })).resolves.toBe(
|
||||
runtimePath
|
||||
)
|
||||
expect(fetcher).not.toHaveBeenCalled()
|
||||
expect(await readFile(runtimePath)).toEqual(Buffer.from(runtime))
|
||||
})
|
||||
|
||||
it('downloads, verifies, extracts, and atomically caches the runtime', async () => {
|
||||
const archive = new TextEncoder().encode('pinned archive')
|
||||
const executable = new TextEncoder().encode('pinned node executable')
|
||||
extraction.executable = executable
|
||||
pin(TARGET, archive, executable)
|
||||
const fetcher = responseFetcher(archive)
|
||||
|
||||
const runtimePath = await materializeCachedNodeRuntime(TARGET, cacheRoot, { fetcher })
|
||||
|
||||
expect(await readFile(runtimePath)).toEqual(Buffer.from(executable))
|
||||
if (process.platform !== 'win32') {
|
||||
expect((await stat(runtimePath)).mode & 0o111).toBe(0o111)
|
||||
}
|
||||
expect(fetcher).toHaveBeenCalledWith(
|
||||
expect.stringContaining(`/v${NODE_RUNTIME_PIN.version}/${originalAssets[TARGET].archive}`),
|
||||
expect.objectContaining({ redirect: 'follow' })
|
||||
)
|
||||
expect(await readdir(runtimeDirFor())).toEqual(['node'])
|
||||
})
|
||||
|
||||
it('refuses an oversized declared archive before reading its body', async () => {
|
||||
const fetcher = responseFetcher(new Uint8Array([1]), 200 * 1024 * 1024 + 1)
|
||||
|
||||
await expect(materializeCachedNodeRuntime(TARGET, cacheRoot, { fetcher })).rejects.toThrow(
|
||||
'Node download exceeded the archive size limit'
|
||||
)
|
||||
await expect(access(join(runtimeDirFor(), 'node'))).rejects.toThrow()
|
||||
expect(await readdir(runtimeDirFor())).toEqual([])
|
||||
})
|
||||
|
||||
it('removes temporary data after an archive checksum mismatch', async () => {
|
||||
const archive = new TextEncoder().encode('tampered archive')
|
||||
const fetcher = responseFetcher(archive)
|
||||
|
||||
await expect(materializeCachedNodeRuntime(TARGET, cacheRoot, { fetcher })).rejects.toThrow(
|
||||
'Node archive checksum mismatch'
|
||||
)
|
||||
expect(await readdir(runtimeDirFor())).toEqual([])
|
||||
})
|
||||
|
||||
it('refuses an executable mismatch even when the archive matches its pin', async () => {
|
||||
const archive = new TextEncoder().encode('pinned archive')
|
||||
extraction.executable = new TextEncoder().encode('incorrect executable')
|
||||
pin(TARGET, archive)
|
||||
await expect(
|
||||
materializeCachedNodeRuntime(TARGET, cacheRoot, { fetcher: responseFetcher(archive) })
|
||||
).rejects.toThrow('Node executable checksum mismatch')
|
||||
expect(await readdir(runtimeDirFor())).toEqual([])
|
||||
})
|
||||
|
||||
// Both cases spawn for real, so the assertion is against the errno Node actually reports:
|
||||
// a missing program rejects asynchronously with ENOENT, while a program path whose parent is a
|
||||
// regular file throws ENOTDIR synchronously out of `spawn` itself.
|
||||
it.each([
|
||||
['absent', (): string => join(cacheRoot, 'absent-extractor')],
|
||||
['unreachable through a file', (): string => join(cacheRoot, 'plain-file', 'unzip')]
|
||||
])('names the %s extractor and the override when it cannot be launched', async (_label, path) => {
|
||||
const target = 'win32-x64' as const
|
||||
const archive = new TextEncoder().encode('unextractable archive')
|
||||
pin(target, archive)
|
||||
await writeFile(join(cacheRoot, 'plain-file'), 'not a directory')
|
||||
const { runProcess: spawnForReal } = await vi.importActual<{
|
||||
runProcess: typeof runProcess
|
||||
}>('@orca/process-host')
|
||||
vi.mocked(runProcess).mockImplementationOnce(spawnForReal)
|
||||
vi.stubEnv('ORCA_UNZIP_BIN', path())
|
||||
|
||||
await expect(
|
||||
materializeCachedNodeRuntime(target, cacheRoot, { fetcher: responseFetcher(archive) })
|
||||
).rejects.toThrow(/install unzip, or set ORCA_UNZIP_BIN/)
|
||||
expect(await readdir(runtimeDirFor(target))).toEqual([])
|
||||
})
|
||||
|
||||
it('cleans an aborted download before publishing any executable', async () => {
|
||||
const controller = new AbortController()
|
||||
controller.abort(new Error('deployment cancelled'))
|
||||
await expect(
|
||||
materializeCachedNodeRuntime(TARGET, cacheRoot, {
|
||||
fetcher: responseFetcher(new Uint8Array([1])),
|
||||
signal: controller.signal
|
||||
})
|
||||
).rejects.toThrow('deployment cancelled')
|
||||
await expect(access(join(cacheRoot, 'node'))).rejects.toThrow()
|
||||
})
|
||||
})
|
||||
|
||||
it('publishes concurrent runtime downloads without removing or replacing the winning executable', async () => {
|
||||
const archive = new TextEncoder().encode('pinned archive')
|
||||
extraction.executable = new TextEncoder().encode('pinned runtime')
|
||||
pin(TARGET, archive, extraction.executable)
|
||||
let finishSecond: (response: Response) => void = () => {}
|
||||
const secondFetcher = vi.fn<typeof fetch>(
|
||||
() =>
|
||||
new Promise<Response>((resolve) => {
|
||||
finishSecond = resolve
|
||||
})
|
||||
)
|
||||
const second = materializeCachedNodeRuntime(TARGET, cacheRoot, { fetcher: secondFetcher })
|
||||
await vi.waitFor(() => expect(secondFetcher).toHaveBeenCalledOnce())
|
||||
const firstPath = await materializeCachedNodeRuntime(TARGET, cacheRoot, {
|
||||
fetcher: responseFetcher(archive)
|
||||
})
|
||||
const firstIdentity = await stat(firstPath)
|
||||
finishSecond(new Response(Buffer.from(archive)))
|
||||
expect(await second).toBe(firstPath)
|
||||
expect((await stat(firstPath)).ino).toBe(firstIdentity.ino)
|
||||
expect(await readFile(firstPath)).toEqual(Buffer.from(extraction.executable))
|
||||
})
|
||||
|
||||
it('repairs a corrupt published runtime beside the old inode and reuses the repair', async () => {
|
||||
const archive = new TextEncoder().encode('pinned archive')
|
||||
extraction.executable = new TextEncoder().encode('pinned runtime')
|
||||
pin(TARGET, archive, extraction.executable)
|
||||
const runtimeDir = runtimeDirFor()
|
||||
const runtimePath = join(runtimeDir, 'node')
|
||||
await mkdir(runtimeDir, { recursive: true })
|
||||
await writeFile(runtimePath, 'corrupt')
|
||||
const fetcher = responseFetcher(archive)
|
||||
const repaired = await materializeCachedNodeRuntime(TARGET, cacheRoot, { fetcher })
|
||||
expect(repaired).not.toBe(runtimePath)
|
||||
expect(await readFile(repaired)).toEqual(Buffer.from(extraction.executable))
|
||||
expect(await materializeCachedNodeRuntime(TARGET, cacheRoot, { fetcher })).toBe(repaired)
|
||||
expect(fetcher).toHaveBeenCalledOnce()
|
||||
expect(await readFile(runtimePath, 'utf8')).toBe('corrupt')
|
||||
})
|
||||
|
||||
it('uses the configured HTTP client for deployment downloads', async () => {
|
||||
const archive = new TextEncoder().encode('proxy archive')
|
||||
extraction.executable = new TextEncoder().encode('proxy runtime')
|
||||
pin(TARGET, archive, extraction.executable)
|
||||
const fetcher = responseFetcher(archive)
|
||||
setMainHttpClient({ fetch: fetcher, proxySession: () => null })
|
||||
await materializeCachedNodeRuntime(TARGET, cacheRoot, {})
|
||||
expect(fetcher).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('allows a progressing download to exceed two minutes', async () => {
|
||||
vi.useFakeTimers()
|
||||
const first = new TextEncoder().encode('first')
|
||||
const second = new TextEncoder().encode('second')
|
||||
extraction.executable = new TextEncoder().encode('slow runtime')
|
||||
pin(TARGET, Buffer.concat([first, second]), extraction.executable)
|
||||
let stream: ReadableStreamDefaultController<Uint8Array> | undefined
|
||||
let signal: AbortSignal | null | undefined
|
||||
const fetcher = vi.fn<typeof fetch>(async (_url, options) => {
|
||||
signal = options?.signal
|
||||
return new Response(
|
||||
new ReadableStream({
|
||||
start(controller) {
|
||||
stream = controller
|
||||
}
|
||||
})
|
||||
)
|
||||
})
|
||||
const pending = materializeCachedNodeRuntime(TARGET, cacheRoot, { fetcher })
|
||||
await vi.waitFor(() => expect(fetcher).toHaveBeenCalledOnce())
|
||||
await vi.advanceTimersByTimeAsync(90_000)
|
||||
stream!.enqueue(first)
|
||||
await vi.waitFor(async () => {
|
||||
const runtimeDir = runtimeDirFor()
|
||||
const temporary = (await readdir(runtimeDir)).find((entry) => entry.startsWith('.download-'))!
|
||||
expect(
|
||||
(await stat(join(runtimeDir, temporary, NODE_RUNTIME_ASSETS[TARGET].archive))).size
|
||||
).toBe(first.length)
|
||||
})
|
||||
await vi.advanceTimersByTimeAsync(90_000)
|
||||
expect(signal?.aborted).toBe(false)
|
||||
stream!.enqueue(second)
|
||||
stream!.close()
|
||||
expect(await readFile(await pending)).toEqual(Buffer.from(extraction.executable))
|
||||
})
|
||||
|
||||
it('retries a stalled body, then gives up and removes the unfinished download', async () => {
|
||||
vi.useFakeTimers()
|
||||
vi.spyOn(console, 'warn').mockImplementation(() => undefined)
|
||||
const cancel = vi.fn()
|
||||
const fetcher = vi.fn<typeof fetch>(async () => new Response(new ReadableStream({ cancel })))
|
||||
const pending = materializeCachedNodeRuntime(TARGET, cacheRoot, { fetcher })
|
||||
const rejected = expect(pending).rejects.toThrow('Node download stalled')
|
||||
const attempts = RUNTIME_ARCHIVE_RETRY_DELAYS_MS.length + 1
|
||||
// Real file I/O sits between the timers, so step the clock until each attempt shows.
|
||||
for (let attempt = 1; attempt <= attempts; attempt += 1) {
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1_000)
|
||||
expect(fetcher).toHaveBeenCalledTimes(attempt)
|
||||
})
|
||||
await vi.waitFor(async () => {
|
||||
await vi.advanceTimersByTimeAsync(10_000)
|
||||
expect(cancel).toHaveBeenCalledTimes(attempt)
|
||||
})
|
||||
}
|
||||
await rejected
|
||||
expect(fetcher).toHaveBeenCalledTimes(attempts)
|
||||
expect(await readdir(runtimeDirFor())).toEqual([])
|
||||
})
|
||||
@@ -0,0 +1,228 @@
|
||||
/** Download, hash-verify and cache the pinned Node runtime, driven only by its pinned asset. */
|
||||
import { createHash, randomUUID } from 'node:crypto'
|
||||
import { createReadStream } from 'node:fs'
|
||||
import { chmod, link, mkdir, rm } from 'node:fs/promises'
|
||||
import { basename, join } from 'node:path'
|
||||
import { getMainHttpClient, type MainHttpClient } from '../network/http-client'
|
||||
import { downloadVerifiedArchive, extractRuntimeArchive } from './runtime-archive-download'
|
||||
import { findOrcadCachePath } from './orcad-cache-path'
|
||||
import { orcadNodeRuntimeExecutable } from '../../shared/orcad-artifacts'
|
||||
import {
|
||||
pinnedNodeRuntimeAsset,
|
||||
nodeRuntimeExecutablePath,
|
||||
nodeRuntimeReleaseUrl,
|
||||
type NodeRuntimeTarget
|
||||
} from '../../shared/node-runtime-pin'
|
||||
|
||||
export type PinnedRuntimeMaterializeOptions = {
|
||||
fetcher?: MainHttpClient['fetch']
|
||||
signal?: AbortSignal
|
||||
}
|
||||
|
||||
export type PinnedRuntimeArchive = {
|
||||
/** Shown in errors, e.g. "Node". */
|
||||
label: string
|
||||
url: string
|
||||
archiveSha256: string
|
||||
}
|
||||
|
||||
export type PinnedRuntimeExecutable = PinnedRuntimeArchive & {
|
||||
executableSha256: string
|
||||
/** Archive-relative path of the executable. */
|
||||
member: string
|
||||
isWindows: boolean
|
||||
/** Where the verified executable is published; `attempt` > 0 names a repair beside it. */
|
||||
cachePath: (attempt: number) => string
|
||||
}
|
||||
|
||||
function archiveName(archive: PinnedRuntimeArchive): string {
|
||||
return basename(new URL(archive.url).pathname)
|
||||
}
|
||||
|
||||
/** The pinned archive, downloaded once and kept by its own name; its hash is checked on every use. */
|
||||
export async function materializeVerifiedRuntimeArchive(
|
||||
archive: PinnedRuntimeArchive,
|
||||
cacheDir: string,
|
||||
options: PinnedRuntimeMaterializeOptions
|
||||
): Promise<string> {
|
||||
options.signal?.throwIfAborted()
|
||||
await mkdir(cacheDir, { recursive: true })
|
||||
const name = archiveName(archive)
|
||||
const cached = await findOrcadCachePath(
|
||||
(attempt) => join(cacheDir, `${attempt ? `repair-${attempt}-` : ''}${name}`),
|
||||
async (path) => (await fileSha256(path)) === archive.archiveSha256
|
||||
)
|
||||
if (cached.verified) {
|
||||
return cached.path
|
||||
}
|
||||
const temporaryDir = join(cacheDir, `.download-${process.pid}-${randomUUID()}`)
|
||||
await mkdir(temporaryDir, { recursive: true })
|
||||
try {
|
||||
const downloaded = join(temporaryDir, name)
|
||||
await downloadVerifiedArchive(
|
||||
archive,
|
||||
downloaded,
|
||||
options.fetcher ?? getMainHttpClient().fetch,
|
||||
options.signal
|
||||
)
|
||||
options.signal?.throwIfAborted()
|
||||
await publishVerified(
|
||||
downloaded,
|
||||
cached.path,
|
||||
archive.archiveSha256,
|
||||
`${archive.label} archive`
|
||||
)
|
||||
return cached.path
|
||||
} finally {
|
||||
await rm(temporaryDir, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
|
||||
export async function materializeCachedRuntimeExecutable(
|
||||
runtime: PinnedRuntimeExecutable,
|
||||
options: PinnedRuntimeMaterializeOptions
|
||||
): Promise<string> {
|
||||
options.signal?.throwIfAborted()
|
||||
const cached = await findOrcadCachePath(
|
||||
runtime.cachePath,
|
||||
async (path) => (await fileSha256(path)) === runtime.executableSha256
|
||||
)
|
||||
const runtimePath = cached.path
|
||||
if (cached.verified) {
|
||||
if (!runtime.isWindows) {
|
||||
await chmod(runtimePath, 0o755)
|
||||
}
|
||||
return runtimePath
|
||||
}
|
||||
const runtimeDir = join(runtimePath, '..')
|
||||
await mkdir(runtimeDir, { recursive: true })
|
||||
const temporaryDir = join(runtimeDir, `.download-${process.pid}-${randomUUID()}`)
|
||||
await mkdir(temporaryDir, { recursive: true })
|
||||
try {
|
||||
const archivePath = join(temporaryDir, archiveName(runtime))
|
||||
await downloadVerifiedArchive(
|
||||
runtime,
|
||||
archivePath,
|
||||
options.fetcher ?? getMainHttpClient().fetch,
|
||||
options.signal
|
||||
)
|
||||
options.signal?.throwIfAborted()
|
||||
const extractedDir = join(temporaryDir, 'extracted')
|
||||
await mkdir(extractedDir)
|
||||
const result = await extractRuntimeArchive(runtime, archivePath, extractedDir, options.signal)
|
||||
options.signal?.throwIfAborted()
|
||||
if (result.code !== 0) {
|
||||
throw new Error(
|
||||
`${runtime.label} archive extraction failed: ${result.stderr || result.stdout}`
|
||||
)
|
||||
}
|
||||
const executable = join(extractedDir, ...runtime.member.split('/'))
|
||||
await verifyFileSha256(executable, runtime.executableSha256, `${runtime.label} executable`)
|
||||
if (!runtime.isWindows) {
|
||||
await chmod(executable, 0o755)
|
||||
}
|
||||
options.signal?.throwIfAborted()
|
||||
await publishVerified(
|
||||
executable,
|
||||
runtimePath,
|
||||
runtime.executableSha256,
|
||||
`cached ${runtime.label} executable`
|
||||
)
|
||||
return runtimePath
|
||||
} finally {
|
||||
await rm(temporaryDir, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
|
||||
/** Hard-link publication: a concurrent winner's identical bytes are kept, never replaced. */
|
||||
async function publishVerified(
|
||||
source: string,
|
||||
destination: string,
|
||||
sha256: string,
|
||||
label: string
|
||||
): Promise<void> {
|
||||
try {
|
||||
await link(source, destination)
|
||||
} catch (error) {
|
||||
if ((await fileSha256(destination)) !== sha256) {
|
||||
throw new Error(`${label} cache entry is unavailable or corrupted: ${destination}`, {
|
||||
cause: error
|
||||
})
|
||||
}
|
||||
}
|
||||
await verifyFileSha256(destination, sha256, label)
|
||||
}
|
||||
|
||||
function nodeRuntimeArchive(target: NodeRuntimeTarget): PinnedRuntimeArchive {
|
||||
const asset = pinnedNodeRuntimeAsset(target)
|
||||
return {
|
||||
label: 'Node',
|
||||
url: nodeRuntimeReleaseUrl(asset.source, asset.archive),
|
||||
archiveSha256: asset.archiveSha256
|
||||
}
|
||||
}
|
||||
|
||||
/** The pinned official archive, uploaded to hosts as published (design D5). */
|
||||
export function materializeNodeRuntimeArchive(
|
||||
target: NodeRuntimeTarget,
|
||||
cacheRoot: string,
|
||||
options: PinnedRuntimeMaterializeOptions
|
||||
): Promise<string> {
|
||||
return materializeVerifiedRuntimeArchive(
|
||||
nodeRuntimeArchive(target),
|
||||
join(cacheRoot, 'node', 'archives'),
|
||||
options
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Client cache keyed by content (design D2): `node/<executableSha256>/node[.exe]`. Only hosts
|
||||
* that cannot extract the archive themselves get the bare executable (Windows SSH vault reads).
|
||||
*/
|
||||
export function materializeCachedNodeRuntime(
|
||||
target: NodeRuntimeTarget,
|
||||
cacheRoot: string,
|
||||
options: PinnedRuntimeMaterializeOptions
|
||||
): Promise<string> {
|
||||
const asset = pinnedNodeRuntimeAsset(target)
|
||||
const executableName = basename(orcadNodeRuntimeExecutable(target))
|
||||
return materializeCachedRuntimeExecutable(
|
||||
{
|
||||
...nodeRuntimeArchive(target),
|
||||
executableSha256: asset.executableSha256,
|
||||
member: nodeRuntimeExecutablePath(target, asset.archive),
|
||||
isWindows: target.startsWith('win32-'),
|
||||
cachePath: (attempt) =>
|
||||
join(
|
||||
cacheRoot,
|
||||
'node',
|
||||
asset.executableSha256,
|
||||
`${attempt ? `repair-${attempt}-` : ''}${executableName}`
|
||||
)
|
||||
},
|
||||
options
|
||||
)
|
||||
}
|
||||
|
||||
export async function verifyFileSha256(
|
||||
path: string,
|
||||
expected: string,
|
||||
label: string
|
||||
): Promise<void> {
|
||||
const actual = await fileSha256(path)
|
||||
if (actual !== expected) {
|
||||
throw new Error(`${label} checksum mismatch: expected ${expected}, got ${actual ?? 'missing'}`)
|
||||
}
|
||||
}
|
||||
|
||||
export async function fileSha256(path: string): Promise<string | null> {
|
||||
try {
|
||||
const hash = createHash('sha256')
|
||||
for await (const chunk of createReadStream(path)) {
|
||||
hash.update(chunk)
|
||||
}
|
||||
return hash.digest('hex')
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,125 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { mkdtemp, readFile, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { PinnedRuntimeArchive } from './pinned-runtime-materializer'
|
||||
import { downloadVerifiedArchive } from './runtime-archive-download'
|
||||
|
||||
const BODY = new Uint8Array([1, 2, 3, 4, 5, 6, 7, 8])
|
||||
const NO_DELAY = [0, 0, 0]
|
||||
let dir = ''
|
||||
let destination = ''
|
||||
|
||||
function archive(body: Uint8Array = BODY): PinnedRuntimeArchive {
|
||||
return {
|
||||
label: 'Node runtime',
|
||||
url: 'https://nodejs.org/dist/node.tar.gz',
|
||||
archiveSha256: createHash('sha256').update(body).digest('hex')
|
||||
}
|
||||
}
|
||||
|
||||
const ok = (body: Uint8Array = BODY): Response => new Response(Buffer.from(body), { status: 200 })
|
||||
|
||||
function requestSignal(options: RequestInit | undefined): AbortSignal {
|
||||
if (!options?.signal) {
|
||||
throw new Error('Expected an owned archive request signal')
|
||||
}
|
||||
return options.signal
|
||||
}
|
||||
|
||||
/** A body that delivers its first half, then fails as Chromium does when the network drops. */
|
||||
function droppedMidway(): Response {
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
controller.enqueue(BODY.subarray(0, 4))
|
||||
},
|
||||
pull(controller) {
|
||||
controller.error(new Error('net::ERR_CONNECTION_RESET'))
|
||||
}
|
||||
})
|
||||
return new Response(stream, { status: 200 })
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
dir = await mkdtemp(join(tmpdir(), 'orca-runtime-archive-'))
|
||||
destination = join(dir, 'node.tar.gz')
|
||||
vi.spyOn(console, 'warn').mockImplementation(() => undefined)
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks()
|
||||
await rm(dir, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe('downloading a pinned runtime archive over a flaky network', () => {
|
||||
it.each([
|
||||
['a network change', () => Promise.reject(new Error('net::ERR_NETWORK_CHANGED'))],
|
||||
['a lost connection', () => Promise.reject(new Error('net::ERR_INTERNET_DISCONNECTED'))],
|
||||
['a server error', () => Promise.resolve(new Response('busy', { status: 503 }))],
|
||||
['a reset partway through the body', () => Promise.resolve(droppedMidway())]
|
||||
])('retries after %s and still verifies the checksum', async (_name, first) => {
|
||||
const signals: AbortSignal[] = []
|
||||
let retiredBeforeRetry = false
|
||||
const fetcher = vi
|
||||
.fn<typeof fetch>()
|
||||
.mockImplementationOnce(async (_url, options) => {
|
||||
signals.push(requestSignal(options))
|
||||
return first()
|
||||
})
|
||||
.mockImplementation(async (_url, options) => {
|
||||
retiredBeforeRetry = signals[0]?.aborted === true
|
||||
signals.push(requestSignal(options))
|
||||
return ok()
|
||||
})
|
||||
await downloadVerifiedArchive(archive(), destination, fetcher, undefined, NO_DELAY)
|
||||
expect(fetcher).toHaveBeenCalledTimes(2)
|
||||
expect(new Uint8Array(await readFile(destination))).toEqual(BODY)
|
||||
expect(retiredBeforeRetry).toBe(true)
|
||||
expect(signals[0]?.aborted).toBe(true)
|
||||
expect(signals[1]?.aborted).toBe(false)
|
||||
})
|
||||
|
||||
it('gives up after its bounded attempts with the last error', async () => {
|
||||
const signals: AbortSignal[] = []
|
||||
const failure = new Error('net::ERR_NETWORK_CHANGED')
|
||||
const fetcher = vi.fn<typeof fetch>(async (_url, options) => {
|
||||
signals.push(requestSignal(options))
|
||||
throw failure
|
||||
})
|
||||
await expect(
|
||||
downloadVerifiedArchive(archive(), destination, fetcher, undefined, NO_DELAY)
|
||||
).rejects.toBe(failure)
|
||||
expect(fetcher).toHaveBeenCalledTimes(NO_DELAY.length + 1)
|
||||
expect(signals).toHaveLength(NO_DELAY.length + 1)
|
||||
expect(signals.every((signal) => signal.aborted)).toBe(true)
|
||||
})
|
||||
|
||||
it.each([
|
||||
['a missing asset', () => new Response('gone', { status: 404 })],
|
||||
['a checksum mismatch', () => ok(new Uint8Array([9, 9]))]
|
||||
])('never retries %s', async (_name, response) => {
|
||||
const signals: AbortSignal[] = []
|
||||
const fetcher = vi.fn<typeof fetch>(async (_url, options) => {
|
||||
signals.push(requestSignal(options))
|
||||
return response()
|
||||
})
|
||||
await expect(
|
||||
downloadVerifiedArchive(archive(), destination, fetcher, undefined, NO_DELAY)
|
||||
).rejects.toThrow()
|
||||
expect(fetcher).toHaveBeenCalledTimes(1)
|
||||
expect(signals[0]?.aborted).toBe(true)
|
||||
})
|
||||
|
||||
it('stops waiting to retry once the caller cancels', async () => {
|
||||
const controller = new AbortController()
|
||||
const fetcher = vi.fn<typeof fetch>(async () => {
|
||||
setTimeout(() => controller.abort(new Error('cancelled')), 10)
|
||||
throw new Error('net::ERR_NETWORK_CHANGED')
|
||||
})
|
||||
await expect(
|
||||
downloadVerifiedArchive(archive(), destination, fetcher, controller.signal, [60_000])
|
||||
).rejects.toThrow('cancelled')
|
||||
expect(fetcher).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,175 @@
|
||||
/** Streams and hash-checks a pinned runtime archive, and extracts it with the host's own tools. */
|
||||
import { createHash } from 'node:crypto'
|
||||
import { open, rm } from 'node:fs/promises'
|
||||
import { runProcess } from '@orca/process-host'
|
||||
import type { ProcessResult } from '@orca/process-host/process-spec'
|
||||
import { waitForPromiseWithSignal } from '../../shared/abort-signal-reason'
|
||||
import { isDefinitiveAbsence } from '../../shared/definitive-filesystem-absence'
|
||||
import { getZipExtractorCommand } from '../../shared/zip-extractor-command'
|
||||
import type { MainHttpClient } from '../network/http-client'
|
||||
import { isRetryableDownloadError, type HttpStatusError } from '../network/transient-download-error'
|
||||
import type { PinnedRuntimeArchive, PinnedRuntimeExecutable } from './pinned-runtime-materializer'
|
||||
|
||||
const MAX_RUNTIME_ARCHIVE_BYTES = 200 * 1024 * 1024
|
||||
/** Pauses before attempts 2..n; a network change or a CDN 5xx usually clears within seconds. */
|
||||
export const RUNTIME_ARCHIVE_RETRY_DELAYS_MS = [1_000, 3_000, 9_000]
|
||||
|
||||
/**
|
||||
* Why the extractor needs a message of its own: `unzip` is absent from a minimal POSIX install,
|
||||
* and a bare `spawn unzip ENOENT` names neither the missing tool nor the override. A misconfigured
|
||||
* `ORCA_UNZIP_BIN` whose parent is a file reports ENOTDIR instead, which is the same verdict.
|
||||
*
|
||||
* The errno is the program path's, not the caller's: `runProcess` leaves cwd unset, so the child
|
||||
* inherits the parent's without resolving it. Measured on macOS, Linux and Windows — spawn still
|
||||
* succeeds from a deleted cwd, even though `process.cwd()` itself throws ENOENT there.
|
||||
*/
|
||||
export async function extractRuntimeArchive(
|
||||
runtime: Pick<PinnedRuntimeExecutable, 'label' | 'member'>,
|
||||
archivePath: string,
|
||||
extractDir: string,
|
||||
signal?: AbortSignal
|
||||
): Promise<ProcessResult> {
|
||||
const { label } = runtime
|
||||
// Why tar for .tar.gz: every POSIX host and Windows 10+ ships it; Node publishes no zip there.
|
||||
const command = archivePath.endsWith('.zip')
|
||||
? getZipExtractorCommand(archivePath, extractDir)
|
||||
: {
|
||||
file: 'tar',
|
||||
args: ['-xzf', archivePath, '-C', extractDir, runtime.member],
|
||||
label: 'tar'
|
||||
}
|
||||
try {
|
||||
return await runProcess({
|
||||
program: command.file,
|
||||
args: command.args,
|
||||
timeoutMs: 120_000,
|
||||
signal
|
||||
})
|
||||
} catch (error) {
|
||||
if (isDefinitiveAbsence(error)) {
|
||||
const override = archivePath.endsWith('.zip')
|
||||
? ', or set ORCA_UNZIP_BIN to an unzip-compatible extractor.'
|
||||
: '.'
|
||||
throw new Error(
|
||||
`${label} archive extraction could not run ${command.file}: install ${command.label}${override}`,
|
||||
{ cause: error }
|
||||
)
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/** Retries transient network failures; a checksum mismatch or a cancel is final. */
|
||||
export async function downloadVerifiedArchive(
|
||||
archive: PinnedRuntimeArchive,
|
||||
destination: string,
|
||||
fetcher: MainHttpClient['fetch'],
|
||||
signal?: AbortSignal,
|
||||
retryDelaysMs: readonly number[] = RUNTIME_ARCHIVE_RETRY_DELAYS_MS
|
||||
): Promise<void> {
|
||||
for (let attempt = 0; ; attempt += 1) {
|
||||
try {
|
||||
await downloadVerifiedArchiveOnce(archive, destination, fetcher, signal)
|
||||
return
|
||||
} catch (error) {
|
||||
const delay = retryDelaysMs[attempt]
|
||||
if (delay === undefined || signal?.aborted || !isRetryableDownloadError(error)) {
|
||||
throw error
|
||||
}
|
||||
console.warn(
|
||||
`[ssh] ${archive.label} download failed; retrying in ${delay}ms (attempt ${attempt + 2}):`,
|
||||
error instanceof Error ? error.message : error
|
||||
)
|
||||
// Why: each attempt opens the destination exclusively, so a partial file must go first.
|
||||
await rm(destination, { force: true })
|
||||
await waitForPromiseWithSignal(new Promise((resolve) => setTimeout(resolve, delay)), signal)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function downloadVerifiedArchiveOnce(
|
||||
archive: PinnedRuntimeArchive,
|
||||
destination: string,
|
||||
fetcher: MainHttpClient['fetch'],
|
||||
signal?: AbortSignal
|
||||
): Promise<void> {
|
||||
const { label, url, archiveSha256: expectedSha256 } = archive
|
||||
const stall = new AbortController()
|
||||
const downloadSignal = signal ? AbortSignal.any([signal, stall.signal]) : stall.signal
|
||||
const stalled: HttpStatusError = Object.assign(new Error(`${label} download stalled`), {
|
||||
retryable: true
|
||||
})
|
||||
const stallTimer = setTimeout(() => stall.abort(stalled), 120_000)
|
||||
try {
|
||||
const response = await fetcher(url, { redirect: 'follow', signal: downloadSignal })
|
||||
if (!response.ok || !response.body) {
|
||||
await response.body?.cancel().catch(() => undefined)
|
||||
throw Object.assign(
|
||||
new Error(`${label} download failed: ${response.status} ${response.statusText}`),
|
||||
{ httpStatusCode: response.status }
|
||||
)
|
||||
}
|
||||
const declaredLength = Number(response.headers.get('content-length'))
|
||||
if (Number.isFinite(declaredLength) && declaredLength > MAX_RUNTIME_ARCHIVE_BYTES) {
|
||||
await response.body.cancel().catch(() => undefined)
|
||||
throw new Error(`${label} download exceeded the archive size limit`)
|
||||
}
|
||||
const handle = await open(destination, 'wx', 0o600).catch(async (error) => {
|
||||
await response.body?.cancel().catch(() => undefined)
|
||||
throw error
|
||||
})
|
||||
const reader = response.body.getReader()
|
||||
const hash = createHash('sha256')
|
||||
let total = 0
|
||||
try {
|
||||
for (;;) {
|
||||
downloadSignal.throwIfAborted()
|
||||
const chunk = await waitForPromiseWithSignal(reader.read(), downloadSignal)
|
||||
if (chunk.done) {
|
||||
break
|
||||
}
|
||||
if (chunk.value.byteLength > 0) {
|
||||
stallTimer.refresh()
|
||||
}
|
||||
total += chunk.value.byteLength
|
||||
if (total > MAX_RUNTIME_ARCHIVE_BYTES) {
|
||||
throw new Error(`${label} download exceeded the archive size limit`)
|
||||
}
|
||||
hash.update(chunk.value)
|
||||
await writeAll(handle, chunk.value)
|
||||
}
|
||||
} catch (error) {
|
||||
await reader.cancel().catch(() => undefined)
|
||||
throw error
|
||||
} finally {
|
||||
await handle.close()
|
||||
}
|
||||
const actual = hash.digest('hex')
|
||||
if (actual !== expectedSha256) {
|
||||
throw new Error(
|
||||
`${label} archive checksum mismatch: expected ${expectedSha256}, got ${actual}`
|
||||
)
|
||||
}
|
||||
} catch (error) {
|
||||
// Why: the fetch may reject a stalled abort with its own AbortError rather than our reason.
|
||||
const downloadError = stall.signal.aborted && !signal?.aborted ? stalled : error
|
||||
stall.abort()
|
||||
throw downloadError
|
||||
} finally {
|
||||
clearTimeout(stallTimer)
|
||||
}
|
||||
}
|
||||
|
||||
async function writeAll(
|
||||
handle: Awaited<ReturnType<typeof open>>,
|
||||
bytes: Uint8Array
|
||||
): Promise<void> {
|
||||
let offset = 0
|
||||
while (offset < bytes.byteLength) {
|
||||
const { bytesWritten } = await handle.write(bytes, offset, bytes.byteLength - offset)
|
||||
if (bytesWritten === 0) {
|
||||
throw new Error('Runtime archive write made no progress')
|
||||
}
|
||||
offset += bytesWritten
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user