fix(ota): stream mirrored archive extraction
This commit is contained in:
DIYgod
parent
8a06a0f821
commit
f176051cd0
|
|
@ -13,10 +13,12 @@
|
|||
"fzstd": "0.1.1",
|
||||
"hono": "4.12.1",
|
||||
"ofetch": "1.5.1",
|
||||
"tar-stream": "2.2.0",
|
||||
"zod": "3.25.76"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@cloudflare/workers-types": "^4.20260405.0",
|
||||
"@types/tar-stream": "3.1.4",
|
||||
"typescript": "catalog:",
|
||||
"vitest": "3.2.4",
|
||||
"wrangler": "4.68.1"
|
||||
|
|
|
|||
|
|
@ -1,9 +1,57 @@
|
|||
import tar from "tar-stream"
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"
|
||||
|
||||
import { extractMirroredFiles } from "../lib/archive"
|
||||
import type { GitHubRequestError } from "../lib/github"
|
||||
import { listPublishedOtaReleases } from "../lib/github"
|
||||
import { IMMUTABLE_ASSET_CACHE_CONTROL, putMirroredFiles } from "../lib/r2"
|
||||
import { mirrorReleaseToStorage } from "../lib/sync"
|
||||
|
||||
vi.mock("fzstd", () => {
|
||||
class Decompress {
|
||||
ondata: (chunk: Uint8Array, final?: boolean) => unknown
|
||||
|
||||
constructor(ondata?: (chunk: Uint8Array, final?: boolean) => unknown) {
|
||||
this.ondata = ondata ?? (() => {})
|
||||
}
|
||||
|
||||
push(chunk: Uint8Array, final = false) {
|
||||
const boundary = Math.max(1, Math.ceil(chunk.length / 2))
|
||||
const firstChunk = chunk.subarray(0, boundary)
|
||||
const secondChunk = chunk.subarray(boundary)
|
||||
|
||||
if (firstChunk.byteLength > 0) {
|
||||
this.ondata(firstChunk, final && secondChunk.byteLength === 0)
|
||||
}
|
||||
|
||||
if (secondChunk.byteLength > 0 || final) {
|
||||
this.ondata(secondChunk, final)
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
return { Decompress }
|
||||
})
|
||||
|
||||
const textEncoder = new TextEncoder()
|
||||
const baseRelease = {
|
||||
schemaVersion: 1,
|
||||
product: "mobile",
|
||||
channel: "production",
|
||||
releaseVersion: "0.4.2",
|
||||
releaseKind: "ota",
|
||||
runtimeVersion: "0.4.1",
|
||||
publishedAt: "2026-04-10T12:00:00Z",
|
||||
git: { tag: "mobile/v0.4.2", commit: "abcdef1234567890" },
|
||||
policy: {
|
||||
storeRequired: false,
|
||||
minSupportedBinaryVersion: "0.4.1",
|
||||
message: null,
|
||||
},
|
||||
} as const
|
||||
|
||||
describe("listPublishedOtaReleases", () => {
|
||||
beforeEach(() => {
|
||||
vi.stubGlobal(
|
||||
|
|
@ -155,6 +203,101 @@ describe("listPublishedOtaReleases", () => {
|
|||
})
|
||||
})
|
||||
|
||||
describe("extractMirroredFiles", () => {
|
||||
it("extracts only referenced files from a tar archive", async () => {
|
||||
const archiveBuffer = await createTarArchive([
|
||||
{
|
||||
name: "bundles/ios-main.js",
|
||||
body: "console.log('ios')",
|
||||
},
|
||||
{
|
||||
name: "bundles/unused.js",
|
||||
body: "console.log('unused')",
|
||||
},
|
||||
])
|
||||
|
||||
const files = await extractMirroredFiles({
|
||||
release: {
|
||||
...baseRelease,
|
||||
platforms: {
|
||||
ios: {
|
||||
launchAsset: {
|
||||
path: "bundles/ios-main.js",
|
||||
sha256: "a".repeat(64),
|
||||
contentType: "application/javascript",
|
||||
},
|
||||
assets: [],
|
||||
},
|
||||
},
|
||||
},
|
||||
archiveBuffer,
|
||||
})
|
||||
|
||||
expect(files).toEqual([
|
||||
{
|
||||
key: "mobile/production/0.4.1/0.4.2/ios/bundles/ios-main.js",
|
||||
body: textEncoder.encode("console.log('ios')"),
|
||||
contentType: "application/javascript",
|
||||
},
|
||||
])
|
||||
})
|
||||
|
||||
it("throws when a referenced archive file is missing", async () => {
|
||||
const archiveBuffer = await createTarArchive([
|
||||
{
|
||||
name: "bundles/unused.js",
|
||||
body: "console.log('unused')",
|
||||
},
|
||||
])
|
||||
|
||||
await expect(
|
||||
extractMirroredFiles({
|
||||
release: {
|
||||
...baseRelease,
|
||||
platforms: {
|
||||
ios: {
|
||||
launchAsset: {
|
||||
path: "bundles/ios-main.js",
|
||||
sha256: "a".repeat(64),
|
||||
contentType: "application/javascript",
|
||||
},
|
||||
assets: [],
|
||||
},
|
||||
},
|
||||
},
|
||||
archiveBuffer,
|
||||
}),
|
||||
).rejects.toThrow('Archive is missing referenced file "bundles/ios-main.js"')
|
||||
})
|
||||
})
|
||||
|
||||
describe("putMirroredFiles", () => {
|
||||
it("forwards content metadata and immutable cache headers to R2", async () => {
|
||||
const bucket = {
|
||||
put: vi.fn(async () => null),
|
||||
} as unknown as R2Bucket
|
||||
|
||||
await putMirroredFiles(bucket, [
|
||||
{
|
||||
key: "mobile/production/0.4.1/0.4.2/ios/bundles/ios-main.js",
|
||||
body: textEncoder.encode("console.log('ios')"),
|
||||
contentType: "application/javascript",
|
||||
},
|
||||
])
|
||||
|
||||
expect(bucket.put).toHaveBeenCalledWith(
|
||||
"mobile/production/0.4.1/0.4.2/ios/bundles/ios-main.js",
|
||||
textEncoder.encode("console.log('ios')"),
|
||||
expect.objectContaining({
|
||||
httpMetadata: expect.objectContaining({
|
||||
contentType: "application/javascript",
|
||||
cacheControl: IMMUTABLE_ASSET_CACHE_CONTROL,
|
||||
}),
|
||||
}),
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
describe("mirrorReleaseToStorage", () => {
|
||||
it("stores mirrored release metadata and latest pointers only for platforms with mirrored payloads", async () => {
|
||||
const kvWrites: string[] = []
|
||||
|
|
@ -176,19 +319,7 @@ describe("mirrorReleaseToStorage", () => {
|
|||
await mirrorReleaseToStorage(
|
||||
{
|
||||
release: {
|
||||
schemaVersion: 1,
|
||||
product: "mobile",
|
||||
channel: "production",
|
||||
releaseVersion: "0.4.2",
|
||||
releaseKind: "ota",
|
||||
runtimeVersion: "0.4.1",
|
||||
publishedAt: "2026-04-10T12:00:00Z",
|
||||
git: { tag: "mobile/v0.4.2", commit: "abcdef1234567890" },
|
||||
policy: {
|
||||
storeRequired: false,
|
||||
minSupportedBinaryVersion: "0.4.1",
|
||||
message: null,
|
||||
},
|
||||
...baseRelease,
|
||||
platforms: {
|
||||
ios: {
|
||||
launchAsset: {
|
||||
|
|
@ -236,3 +367,63 @@ describe("mirrorReleaseToStorage", () => {
|
|||
)
|
||||
})
|
||||
})
|
||||
|
||||
async function createTarArchive(
|
||||
entries: Array<{
|
||||
name: string
|
||||
body: string
|
||||
}>,
|
||||
) {
|
||||
const pack = tar.pack()
|
||||
const archiveChunks: Uint8Array[] = []
|
||||
|
||||
const archivePromise = new Promise<Uint8Array>((resolve, reject) => {
|
||||
pack.on("data", (chunk) => {
|
||||
archiveChunks.push(new Uint8Array(chunk))
|
||||
})
|
||||
pack.on("error", reject)
|
||||
pack.on("end", () => {
|
||||
resolve(concatenateChunks(archiveChunks))
|
||||
})
|
||||
})
|
||||
|
||||
for (const entry of entries) {
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const body = textEncoder.encode(entry.body)
|
||||
const tarEntry = pack.entry(
|
||||
{
|
||||
name: entry.name,
|
||||
size: body.byteLength,
|
||||
},
|
||||
(error) => {
|
||||
if (error) {
|
||||
reject(error)
|
||||
return
|
||||
}
|
||||
|
||||
resolve()
|
||||
},
|
||||
)
|
||||
|
||||
tarEntry.on("error", reject)
|
||||
tarEntry.end(body)
|
||||
})
|
||||
}
|
||||
|
||||
pack.finalize()
|
||||
|
||||
return archivePromise
|
||||
}
|
||||
|
||||
function concatenateChunks(chunks: readonly Uint8Array[]) {
|
||||
const totalLength = chunks.reduce((length, chunk) => length + chunk.byteLength, 0)
|
||||
const output = new Uint8Array(totalLength)
|
||||
let offset = 0
|
||||
|
||||
for (const chunk of chunks) {
|
||||
output.set(chunk, offset)
|
||||
offset += chunk.byteLength
|
||||
}
|
||||
|
||||
return output
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,14 +1,20 @@
|
|||
import { decompress } from "fzstd"
|
||||
import { Decompress } from "fzstd"
|
||||
import tar from "tar-stream"
|
||||
|
||||
import type { OtaPlatform, OtaRelease } from "./schema"
|
||||
|
||||
const TAR_BLOCK_SIZE = 512
|
||||
const textDecoder = new TextDecoder()
|
||||
const OTA_PLATFORMS: OtaPlatform[] = ["ios", "android", "macos", "windows", "linux"]
|
||||
|
||||
type PlatformPayload = NonNullable<OtaRelease["platforms"][OtaPlatform]>
|
||||
type PlatformAsset = PlatformPayload["launchAsset"] | PlatformPayload["assets"][number]
|
||||
|
||||
interface MirroredFileRequest {
|
||||
archivePath: string
|
||||
key: string
|
||||
contentType: string
|
||||
body?: Uint8Array
|
||||
}
|
||||
|
||||
export interface MirroredFile {
|
||||
key: string
|
||||
body: Uint8Array
|
||||
|
|
@ -36,35 +42,120 @@ export async function extractMirroredFiles(input: {
|
|||
}): Promise<MirroredFile[]> {
|
||||
const compressedArchive =
|
||||
input.archiveBuffer instanceof Uint8Array ? input.archiveBuffer : new Uint8Array(input.archiveBuffer)
|
||||
const archiveEntries = parseTarArchive(decompress(compressedArchive))
|
||||
const files: MirroredFile[] = []
|
||||
const requestedFiles = createMirroredFileRequests(input.release)
|
||||
const requestsByArchivePath = new Map<string, MirroredFileRequest[]>()
|
||||
|
||||
for (const request of requestedFiles) {
|
||||
const requestsForPath = requestsByArchivePath.get(request.archivePath)
|
||||
|
||||
if (requestsForPath) {
|
||||
requestsForPath.push(request)
|
||||
continue
|
||||
}
|
||||
|
||||
requestsByArchivePath.set(request.archivePath, [request])
|
||||
}
|
||||
|
||||
const missingArchivePaths = new Set(requestsByArchivePath.keys())
|
||||
const tarExtract = tar.extract()
|
||||
|
||||
const extractedFiles = await new Promise<MirroredFile[]>((resolve, reject) => {
|
||||
const rejectOnce = once(reject)
|
||||
const resolveOnce = once(resolve)
|
||||
|
||||
tarExtract.on("entry", (header, stream, next) => {
|
||||
const archivePath = normalizeArchivePath(header.name)
|
||||
const matchingRequests = requestsByArchivePath.get(archivePath)
|
||||
const chunks: Uint8Array[] = []
|
||||
|
||||
stream.on("data", (chunk) => {
|
||||
if (matchingRequests) {
|
||||
chunks.push(new Uint8Array(chunk))
|
||||
}
|
||||
})
|
||||
stream.on("error", (error) => {
|
||||
rejectOnce(toError(error))
|
||||
})
|
||||
stream.on("end", () => {
|
||||
if (matchingRequests) {
|
||||
const body = concatenateChunks(chunks)
|
||||
|
||||
for (const request of matchingRequests) {
|
||||
request.body = body
|
||||
}
|
||||
|
||||
missingArchivePaths.delete(archivePath)
|
||||
}
|
||||
|
||||
next()
|
||||
})
|
||||
stream.resume()
|
||||
})
|
||||
|
||||
tarExtract.on("error", (error) => {
|
||||
rejectOnce(toError(error))
|
||||
})
|
||||
tarExtract.on("finish", () => {
|
||||
if (missingArchivePaths.size > 0) {
|
||||
rejectOnce(
|
||||
new Error(
|
||||
`Archive is missing referenced file "${[...missingArchivePaths][0]}" for ${input.release.releaseVersion}`,
|
||||
),
|
||||
)
|
||||
return
|
||||
}
|
||||
|
||||
resolveOnce(
|
||||
requestedFiles.map((request) => ({
|
||||
key: request.key,
|
||||
body: request.body ?? new Uint8Array(0),
|
||||
contentType: request.contentType,
|
||||
})),
|
||||
)
|
||||
})
|
||||
|
||||
const zstdStream = new Decompress((chunk, final) => {
|
||||
if (chunk.byteLength > 0) {
|
||||
tarExtract.write(chunk)
|
||||
}
|
||||
|
||||
if (final) {
|
||||
tarExtract.end()
|
||||
}
|
||||
})
|
||||
|
||||
try {
|
||||
zstdStream.push(compressedArchive, true)
|
||||
} catch (error) {
|
||||
rejectOnce(toError(error))
|
||||
}
|
||||
})
|
||||
|
||||
return extractedFiles
|
||||
}
|
||||
|
||||
function createMirroredFileRequests(release: OtaRelease): MirroredFileRequest[] {
|
||||
const requests: MirroredFileRequest[] = []
|
||||
|
||||
for (const platform of OTA_PLATFORMS) {
|
||||
const platformPayload = input.release.platforms[platform]
|
||||
const platformPayload = release.platforms[platform]
|
||||
|
||||
if (!platformPayload) {
|
||||
continue
|
||||
}
|
||||
|
||||
for (const asset of listReferencedAssets(platformPayload)) {
|
||||
const assetPath = normalizeArchivePath(asset.path)
|
||||
const archiveBody = archiveEntries.get(assetPath)
|
||||
const archivePath = normalizeArchivePath(asset.path)
|
||||
|
||||
if (!archiveBody) {
|
||||
throw new Error(
|
||||
`Archive is missing referenced file "${assetPath}" for ${platform} in ${input.release.releaseVersion}`,
|
||||
)
|
||||
}
|
||||
|
||||
files.push({
|
||||
key: buildMirroredAssetKey(input.release, platform, assetPath),
|
||||
body: archiveBody,
|
||||
requests.push({
|
||||
archivePath,
|
||||
key: buildMirroredAssetKey(release, platform, archivePath),
|
||||
contentType: asset.contentType,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
return files
|
||||
return requests
|
||||
}
|
||||
|
||||
function listReferencedAssets(platformPayload: PlatformPayload): PlatformAsset[] {
|
||||
|
|
@ -77,75 +168,44 @@ function listReferencedAssets(platformPayload: PlatformPayload): PlatformAsset[]
|
|||
return [...dedupedAssets.values()]
|
||||
}
|
||||
|
||||
function parseTarArchive(archive: Uint8Array) {
|
||||
const entries = new Map<string, Uint8Array>()
|
||||
let offset = 0
|
||||
|
||||
while (offset + TAR_BLOCK_SIZE <= archive.length) {
|
||||
const header = archive.subarray(offset, offset + TAR_BLOCK_SIZE)
|
||||
|
||||
if (isZeroBlock(header)) {
|
||||
break
|
||||
}
|
||||
|
||||
const fileName = readTarPath(header)
|
||||
const fileSize = parseTarOctal(header.subarray(124, 136))
|
||||
const fileType = header[156] ?? 0
|
||||
const fileStart = offset + TAR_BLOCK_SIZE
|
||||
const fileEnd = fileStart + fileSize
|
||||
|
||||
if (fileEnd > archive.length) {
|
||||
throw new Error(`Archive entry "${fileName}" exceeds archive bounds`)
|
||||
}
|
||||
|
||||
if (isRegularFile(fileType)) {
|
||||
entries.set(normalizeArchivePath(fileName), archive.slice(fileStart, fileEnd))
|
||||
}
|
||||
|
||||
offset = fileStart + Math.ceil(fileSize / TAR_BLOCK_SIZE) * TAR_BLOCK_SIZE
|
||||
}
|
||||
|
||||
return entries
|
||||
}
|
||||
|
||||
function readTarPath(header: Uint8Array) {
|
||||
const name = readTarString(header.subarray(0, 100))
|
||||
const prefix = readTarString(header.subarray(345, 500))
|
||||
|
||||
return prefix ? `${prefix}/${name}` : name
|
||||
}
|
||||
|
||||
function readTarString(value: Uint8Array) {
|
||||
const terminatorIndex = value.indexOf(0)
|
||||
const sliceEnd = terminatorIndex === -1 ? value.length : terminatorIndex
|
||||
|
||||
return textDecoder.decode(value.subarray(0, sliceEnd)).trim()
|
||||
}
|
||||
|
||||
function parseTarOctal(value: Uint8Array) {
|
||||
const rawValue = readTarString(value).replaceAll("\0", "").trim()
|
||||
|
||||
if (!rawValue) {
|
||||
return 0
|
||||
}
|
||||
|
||||
const parsedValue = Number.parseInt(rawValue, 8)
|
||||
|
||||
if (Number.isNaN(parsedValue)) {
|
||||
throw new TypeError(`Invalid tar entry size "${rawValue}"`)
|
||||
}
|
||||
|
||||
return parsedValue
|
||||
}
|
||||
|
||||
function normalizeArchivePath(path: string) {
|
||||
return path.replace(/^\/+/, "").replace(/^(\.\/)+/, "").replaceAll(/\/{2,}/g, "/")
|
||||
}
|
||||
|
||||
function isRegularFile(fileType: number) {
|
||||
return fileType === 0 || fileType === 48
|
||||
function concatenateChunks(chunks: readonly Uint8Array[]) {
|
||||
if (chunks.length === 0) {
|
||||
return new Uint8Array(0)
|
||||
}
|
||||
|
||||
if (chunks.length === 1) {
|
||||
return chunks[0]!.slice()
|
||||
}
|
||||
|
||||
const totalLength = chunks.reduce((length, chunk) => length + chunk.byteLength, 0)
|
||||
const output = new Uint8Array(totalLength)
|
||||
let offset = 0
|
||||
|
||||
for (const chunk of chunks) {
|
||||
output.set(chunk, offset)
|
||||
offset += chunk.byteLength
|
||||
}
|
||||
|
||||
return output
|
||||
}
|
||||
|
||||
function isZeroBlock(block: Uint8Array) {
|
||||
return block.every((byte) => byte === 0)
|
||||
function once<T extends (...args: never[]) => void>(callback: T): T {
|
||||
let called = false
|
||||
|
||||
return ((...args: Parameters<T>) => {
|
||||
if (called) {
|
||||
return
|
||||
}
|
||||
|
||||
called = true
|
||||
callback(...args)
|
||||
}) as T
|
||||
}
|
||||
|
||||
function toError(error: unknown) {
|
||||
return error instanceof Error ? error : new Error(String(error))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,9 +1,12 @@
|
|||
import type { MirroredFile } from "./archive"
|
||||
|
||||
export const IMMUTABLE_ASSET_CACHE_CONTROL = "public, max-age=31536000, immutable"
|
||||
|
||||
export async function putMirroredFiles(bucket: R2Bucket, files: readonly MirroredFile[]) {
|
||||
for (const file of files) {
|
||||
await bucket.put(file.key, file.body, {
|
||||
httpMetadata: {
|
||||
cacheControl: IMMUTABLE_ASSET_CACHE_CONTROL,
|
||||
contentType: file.contentType,
|
||||
},
|
||||
})
|
||||
|
|
|
|||
|
|
@ -1453,6 +1453,9 @@ importers:
|
|||
ofetch:
|
||||
specifier: 1.5.1
|
||||
version: 1.5.1
|
||||
tar-stream:
|
||||
specifier: 2.2.0
|
||||
version: 2.2.0
|
||||
zod:
|
||||
specifier: 3.25.76
|
||||
version: 3.25.76
|
||||
|
|
@ -1460,6 +1463,9 @@ importers:
|
|||
'@cloudflare/workers-types':
|
||||
specifier: ^4.20260405.0
|
||||
version: 4.20260410.1
|
||||
'@types/tar-stream':
|
||||
specifier: 3.1.4
|
||||
version: 3.1.4
|
||||
typescript:
|
||||
specifier: 'catalog:'
|
||||
version: 5.9.3
|
||||
|
|
@ -7875,6 +7881,9 @@ packages:
|
|||
'@types/stack-utils@2.0.3':
|
||||
resolution: {integrity: sha512-9aEbYZ3TbYMznPdcdr3SmIrLXwC/AKZXQeCf9Pgao5CKb8CyHuEX5jzWPTkvregvhRJHcpRO6BFoGW9ycaOkYw==}
|
||||
|
||||
'@types/tar-stream@3.1.4':
|
||||
resolution: {integrity: sha512-921gW0+g29mCJX0fRvqeHzBlE/XclDaAG0Ousy1LCghsOhvaKacDeRGEVzQP9IPfKn8Vysy7FEXAIxycpc/CMg==}
|
||||
|
||||
'@types/tar@6.1.13':
|
||||
resolution: {integrity: sha512-IznnlmU5f4WcGTh2ltRu/Ijpmk8wiWXfF0VA4s+HPjHZgvFggk1YaIkbo5krX/zUCzWF8N/l4+W/LNxnvAJ8nw==}
|
||||
|
||||
|
|
@ -24817,6 +24826,10 @@ snapshots:
|
|||
|
||||
'@types/stack-utils@2.0.3': {}
|
||||
|
||||
'@types/tar-stream@3.1.4':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
|
||||
'@types/tar@6.1.13':
|
||||
dependencies:
|
||||
'@types/node': 25.2.3
|
||||
|
|
|
|||
Loading…
Reference in New Issue