mirror of
https://github.com/stablyai/orca.git
synced 2026-11-13 17:37:40 +00:00
perf(plugins): read catalog images in batched Git calls (#38463)
* perf(plugins): read catalog images in batched Git calls Each catalog refresh ran one ls-tree and one cat-file per image or manifest read. Batch every file an extraction round asks for into one ls-tree listing and one cat-file --batch read, then rerun extraction over the loaded files until nothing is missing. * fix(plugins): bound speculative catalog blob prefetch --------- Co-authored-by: m4air <m4air@Mac.localdomain> Co-authored-by: Neil <neil@stably.ai>
This commit is contained in:
@@ -1,11 +1,18 @@
|
||||
import { createHash } from 'node:crypto'
|
||||
import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, expect, it, vi } from 'vitest'
|
||||
import { gitExecFileAsync } from '../../git/runner'
|
||||
import type * as GitRunner from '../../git/runner'
|
||||
import { gitExecFileAsync, gitExecFileAsyncBuffer } from '../../git/runner'
|
||||
import { serveRepositoriesOverTestSsh } from '../__mocks__/plugin-git-ssh-remote-fixture'
|
||||
import { fetchPluginMarketplace } from './plugin-marketplace-fetch'
|
||||
|
||||
vi.mock('../../git/runner', async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof GitRunner>()
|
||||
return { ...actual, gitExecFileAsyncBuffer: vi.fn(actual.gitExecFileAsyncBuffer) }
|
||||
})
|
||||
|
||||
const roots: string[] = []
|
||||
afterEach(async () => {
|
||||
vi.unstubAllEnvs()
|
||||
@@ -51,3 +58,164 @@ it('reads only the catalog index, prefers orca-catalog.json, and ignores plugin
|
||||
expect(fetched.marketplace.name).toBe('New')
|
||||
expect(fetched.marketplaceCommit).toBe((await git(['rev-parse', 'HEAD'])).trim())
|
||||
})
|
||||
|
||||
function png(width: number, height: number, padding = 0): Buffer {
|
||||
const header = Buffer.alloc(33)
|
||||
Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 0, 0, 0, 13]).copy(header)
|
||||
header.write('IHDR', 12, 'ascii')
|
||||
header.writeUInt32BE(width, 16)
|
||||
header.writeUInt32BE(height, 20)
|
||||
header.set([8, 6, 0, 0, 0], 24)
|
||||
return Buffer.concat([header, Buffer.alloc(padding)])
|
||||
}
|
||||
|
||||
it('reads every monorepo plugin image in shared Git calls and refuses oversized files unread', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-marketplace-images-'))
|
||||
roots.push(root)
|
||||
const repository = join(root, 'catalog')
|
||||
const url = 'ssh://git@catalog.example.invalid/catalog.git'
|
||||
const git = async (args: string[]) => (await gitExecFileAsync(args, { cwd: repository })).stdout
|
||||
const plugins = ['alpha', 'beta', 'gamma']
|
||||
for (const [index, name] of plugins.entries()) {
|
||||
await mkdir(join(repository, 'plugins', name, 'images'), { recursive: true })
|
||||
const manifest = {
|
||||
name,
|
||||
cover: { path: 'images/cover.png', alt: name },
|
||||
screenshots: [{ path: 'images/missing.png', alt: 'gone' }]
|
||||
}
|
||||
await writeFile(join(repository, 'plugins', name, 'orca-plugin.json'), JSON.stringify(manifest))
|
||||
// gamma's cover is over the 1 MB cover limit.
|
||||
await writeFile(
|
||||
join(repository, 'plugins', name, 'images', 'cover.png'),
|
||||
png(1600 + index, 1000, name === 'gamma' ? 1024 * 1024 : 0)
|
||||
)
|
||||
}
|
||||
const release = {
|
||||
version: '1.0.0',
|
||||
commit: 'c'.repeat(40),
|
||||
contentHash: 'f'.repeat(64),
|
||||
execution: 'data',
|
||||
minimumOrca: '1.4.245'
|
||||
}
|
||||
await writeFile(
|
||||
join(repository, 'orca-catalog.json'),
|
||||
JSON.stringify({
|
||||
catalogVersion: 2,
|
||||
name: 'Images',
|
||||
owner: 'samples',
|
||||
plugins: plugins.map((name) => ({
|
||||
id: `samples.${name}`,
|
||||
source: { kind: 'git', url, ref: 'main', path: `plugins/${name}` },
|
||||
categories: [],
|
||||
releases: [release]
|
||||
}))
|
||||
})
|
||||
)
|
||||
await git(['init', '--quiet'])
|
||||
await git(['config', 'maintenance.auto', 'false'])
|
||||
await git(['add', '--all'])
|
||||
await git(['-c', 'user.name=t', '-c', 'user.email=t@example.invalid', 'commit', '-qm', 'c'])
|
||||
await git(['tag', 'catalog'])
|
||||
await serveRepositoriesOverTestSsh(root, { '/catalog.git': repository })
|
||||
vi.mocked(gitExecFileAsyncBuffer).mockClear()
|
||||
|
||||
const fetched = await fetchPluginMarketplace({
|
||||
id: 'b'.repeat(32),
|
||||
source: { kind: 'git', url, ref: 'catalog' },
|
||||
addedAt: 0
|
||||
})
|
||||
|
||||
expect(
|
||||
fetched.marketplace.plugins.map((plugin) => plugin.releases?.[0]?.images?.cover?.width ?? null)
|
||||
).toEqual([1600, 1601, null])
|
||||
expect(fetched.images?.map((image) => image.byteLength)).toEqual([33, 33])
|
||||
// Index listing and read, then one listing and one batch read per dependency round.
|
||||
expect(vi.mocked(gitExecFileAsyncBuffer).mock.calls.map(([args]) => args[0])).toEqual([
|
||||
'ls-tree',
|
||||
'cat-file',
|
||||
'ls-tree',
|
||||
'cat-file',
|
||||
'ls-tree',
|
||||
'cat-file'
|
||||
])
|
||||
})
|
||||
|
||||
it.each([7, 70])(
|
||||
'bounds speculative image reads for a %i MB catalog and preserves the verified-image budget',
|
||||
async (count) => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-marketplace-budget-'))
|
||||
roots.push(root)
|
||||
const repository = join(root, 'catalog')
|
||||
const git = async (args: string[]) => (await gitExecFileAsync(args, { cwd: repository })).stdout
|
||||
await mkdir(join(repository, 'images'), { recursive: true })
|
||||
const plugins: unknown[] = []
|
||||
for (let index = 0; index < count; index += 1) {
|
||||
const path = `images/${index}.png`
|
||||
const bytes = png(1600 + index, 1000, 1024 * 1024 - 33)
|
||||
await writeFile(join(repository, path), bytes)
|
||||
plugins.push({
|
||||
id: `samples.image${index}`,
|
||||
source: { kind: 'git', url: 'https://example.com/plugin.git', ref: 'main' },
|
||||
categories: [],
|
||||
releases: [
|
||||
{
|
||||
version: '1.0.0',
|
||||
commit: 'c'.repeat(40),
|
||||
contentHash: 'f'.repeat(64),
|
||||
execution: 'data',
|
||||
minimumOrca: '1.4.245',
|
||||
images: {
|
||||
cover: {
|
||||
path,
|
||||
alt: 'cover',
|
||||
width: 1600 + index,
|
||||
height: 1000,
|
||||
sha256:
|
||||
index === 0 ? '0'.repeat(64) : createHash('sha256').update(bytes).digest('hex')
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
})
|
||||
}
|
||||
await writeFile(
|
||||
join(repository, 'orca-catalog.json'),
|
||||
JSON.stringify({
|
||||
catalogVersion: 2,
|
||||
name: 'Budget',
|
||||
owner: 'samples',
|
||||
plugins
|
||||
})
|
||||
)
|
||||
await git(['init', '--quiet'])
|
||||
await git(['config', 'maintenance.auto', 'false'])
|
||||
await git(['add', '--all'])
|
||||
await git(['-c', 'user.name=t', '-c', 'user.email=t@example.invalid', 'commit', '-qm', 'c'])
|
||||
await git(['tag', 'catalog'])
|
||||
await serveRepositoriesOverTestSsh(root, { '/catalog.git': repository })
|
||||
vi.mocked(gitExecFileAsyncBuffer).mockClear()
|
||||
|
||||
const fetched = await fetchPluginMarketplace({
|
||||
id: 'd'.repeat(32),
|
||||
source: { kind: 'git', url: 'ssh://git@catalog.example.invalid/catalog.git', ref: 'catalog' },
|
||||
addedAt: 0
|
||||
})
|
||||
|
||||
const kept = fetched.marketplace.plugins.flatMap((plugin) => {
|
||||
const cover = plugin.releases?.[0]?.images?.cover
|
||||
return cover ? [cover.width] : []
|
||||
})
|
||||
expect(kept).toEqual(
|
||||
Array.from({ length: Math.min(count - 1, 64) }, (_, index) => 1601 + index)
|
||||
)
|
||||
expect(fetched.images?.reduce((sum, image) => sum + image.byteLength, 0)).toBe(
|
||||
Math.min(count - 1, 64) * 1024 * 1024
|
||||
)
|
||||
const batches = vi
|
||||
.mocked(gitExecFileAsyncBuffer)
|
||||
.mock.calls.filter(([args]) => args[0] === 'cat-file' && args[1] === '--batch')
|
||||
expect(batches.map(([, options]) => options?.stdin?.trim().split('\n').length)).toEqual(
|
||||
count <= 64 ? [count] : Array.from({ length: 65 }, () => 1)
|
||||
)
|
||||
}
|
||||
)
|
||||
|
||||
@@ -11,15 +11,17 @@ import {
|
||||
} from '../../../shared/plugins/catalog/plugin-catalog'
|
||||
import { formatPluginSchemaIssues } from '../../../shared/plugins/manifest/plugin-schema-issues'
|
||||
import { gitExecFileAsyncBuffer } from '../../git/runner'
|
||||
import { extractCatalogImages } from './plugin-catalog-image-extraction'
|
||||
import { extractCatalogImages, type CatalogBlobReader } from './plugin-catalog-image-extraction'
|
||||
import { fillMonorepoReleaseImages } from './plugin-catalog-monorepo-images'
|
||||
import { readPluginGitCommit } from '../installation/plugin-git-repository'
|
||||
import { readPluginGitBatchBlobs, type PluginGitTreeEntry } from '../installation/plugin-git-tree'
|
||||
import type { PluginMarketplaceRegisteredSource } from './plugin-marketplace-store'
|
||||
|
||||
const MARKETPLACE_INDEX_MAX_BYTES = 16 * 1024 * 1024
|
||||
// Why its own cap: a catalog repository may hold far more than one plugin, and only its index is read.
|
||||
const MARKETPLACE_DOWNLOAD_MAX_BYTES = 512 * 1024 * 1024
|
||||
const MARKETPLACE_GIT_TIMEOUT_MS = 120_000
|
||||
const CATALOG_BLOB_CACHE_MAX_BYTES = 64 * 1024 * 1024
|
||||
|
||||
export type PluginMarketplaceFetchResult = {
|
||||
marketplaceCommit: string
|
||||
@@ -77,38 +79,150 @@ async function readMarketplaceAtCommit(
|
||||
...options,
|
||||
maxBuffer: MARKETPLACE_INDEX_MAX_BYTES
|
||||
})
|
||||
const readBlob = (path: string, maxBytes: number): Promise<Buffer | null> =>
|
||||
readCatalogBlob(repository, commit, path, maxBytes)
|
||||
return extractCatalogImages(
|
||||
parsePluginMarketplaceIndex(filename, bytes.stdout),
|
||||
readBlob,
|
||||
(entry, read) => fillMonorepoReleaseImages(entry, catalogUrl, read)
|
||||
const index = parsePluginMarketplaceIndex(filename, bytes.stdout)
|
||||
const batched = await extractBatchedMarketplaceImages(index, repository, commit, catalogUrl)
|
||||
if (batched) {
|
||||
return batched
|
||||
}
|
||||
const readBlob: CatalogBlobReader = async (path, maxBytes) => {
|
||||
const files = new Map<string, CatalogFile>()
|
||||
await loadCatalogFiles(repository, commit, new Map([[path, maxBytes]]), files)
|
||||
return files.get(path)?.bytes ?? null
|
||||
}
|
||||
return extractCatalogImages(index, readBlob, (entry, read) =>
|
||||
fillMonorepoReleaseImages(entry, catalogUrl, read)
|
||||
)
|
||||
}
|
||||
|
||||
/** One file from the catalog commit, refused before reading when it is larger than allowed. */
|
||||
async function readCatalogBlob(
|
||||
async function extractBatchedMarketplaceImages(
|
||||
index: PluginMarketplace,
|
||||
repository: string,
|
||||
commit: string,
|
||||
path: string,
|
||||
maxBytes: number
|
||||
): Promise<Buffer | null> {
|
||||
const options = { cwd: repository, timeout: MARKETPLACE_GIT_TIMEOUT_MS }
|
||||
const listing = await gitExecFileAsyncBuffer(['ls-tree', '-l', '-z', commit, '--', path], {
|
||||
...options,
|
||||
maxBuffer: 4 * 1024
|
||||
})
|
||||
const match = /^100(?:644|755) blob ([0-9a-f]{40}) +([0-9]+)\t/.exec(
|
||||
listing.stdout.toString('utf8')
|
||||
)
|
||||
if (!match?.[1] || Number(match[2]) > maxBytes) {
|
||||
return null
|
||||
catalogUrl: string
|
||||
): Promise<{ marketplace: PluginMarketplace; images: Buffer[] } | null> {
|
||||
const files = new Map<string, CatalogFile>()
|
||||
// Rerun extraction until every file it asks for is loaded, so each round's misses share two Git calls.
|
||||
for (;;) {
|
||||
const missing = new Map<string, number>()
|
||||
const readBlob = async (path: string, maxBytes: number): Promise<Buffer | null> => {
|
||||
const file = files.get(path)
|
||||
if (file === undefined || (file && !file.bytes && file.size <= maxBytes)) {
|
||||
missing.set(path, Math.max(missing.get(path) ?? 0, maxBytes))
|
||||
return null
|
||||
}
|
||||
return file?.bytes && file.size <= maxBytes ? file.bytes : null
|
||||
}
|
||||
const extracted = await extractCatalogImages(index, readBlob, (entry, read) =>
|
||||
fillMonorepoReleaseImages(entry, catalogUrl, read)
|
||||
)
|
||||
if (missing.size === 0) {
|
||||
return extracted
|
||||
}
|
||||
if (!(await loadCatalogFiles(repository, commit, missing, files))) {
|
||||
return null
|
||||
}
|
||||
}
|
||||
const blob = await gitExecFileAsyncBuffer(['cat-file', 'blob', match[1]], {
|
||||
...options,
|
||||
maxBuffer: maxBytes
|
||||
})
|
||||
return blob.stdout
|
||||
}
|
||||
|
||||
/** A regular file's size and, when it fit the largest requested limit, its bytes; null when absent. */
|
||||
type CatalogFile = { size: number; bytes: Buffer | null } | null
|
||||
|
||||
// Keep each ls-tree argv well under the Windows command-line limit.
|
||||
const LS_TREE_PATH_CHUNK_CHARS = 16 * 1024
|
||||
|
||||
/** Loads files from the catalog commit, refusing to read any larger than its requested limit. */
|
||||
async function loadCatalogFiles(
|
||||
repository: string,
|
||||
commit: string,
|
||||
requests: ReadonlyMap<string, number>,
|
||||
files: Map<string, CatalogFile>
|
||||
): Promise<boolean> {
|
||||
const options = { cwd: repository, timeout: MARKETPLACE_GIT_TIMEOUT_MS }
|
||||
const reads: Extract<PluginGitTreeEntry, { kind: 'blob' }>[] = []
|
||||
let cachedBytes = 0
|
||||
for (const file of files.values()) {
|
||||
cachedBytes += file?.bytes?.byteLength ?? 0
|
||||
}
|
||||
for (const chunk of chunkByLength([...requests.keys()], LS_TREE_PATH_CHUNK_CHARS)) {
|
||||
try {
|
||||
const listing = await gitExecFileAsyncBuffer(
|
||||
['ls-tree', '-l', '-z', commit, '--', ...chunk],
|
||||
{
|
||||
...options,
|
||||
maxBuffer: chunk.length * 4 * 1024
|
||||
}
|
||||
)
|
||||
const listed = new Map<string, { object: string; size: number }>()
|
||||
for (const record of listing.stdout.toString('utf8').split('\0')) {
|
||||
const match = /^100(?:644|755) blob ([0-9a-f]{40}) +([0-9]+)\t(.+)$/s.exec(record)
|
||||
if (match) {
|
||||
listed.set(match[3]!, { object: match[1]!, size: Number(match[2]) })
|
||||
}
|
||||
}
|
||||
for (const path of chunk) {
|
||||
const found = listed.get(gitTreePath(path))
|
||||
if (found && found.size <= requests.get(path)!) {
|
||||
cachedBytes += found.size
|
||||
if (cachedBytes > CATALOG_BLOB_CACHE_MAX_BYTES) {
|
||||
// Fall back before reading this round; extraction's own budget controls sequential reads.
|
||||
return false
|
||||
}
|
||||
reads.push({
|
||||
kind: 'blob',
|
||||
path,
|
||||
object: found.object,
|
||||
size: found.size,
|
||||
executable: false
|
||||
})
|
||||
} else {
|
||||
files.set(path, found ? { size: found.size, bytes: null } : null)
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
for (const path of chunk) {
|
||||
files.set(path, null)
|
||||
}
|
||||
}
|
||||
}
|
||||
if (reads.length === 0) {
|
||||
return true
|
||||
}
|
||||
try {
|
||||
const objects = [...new Map(reads.map((read) => [read.object, read])).values()]
|
||||
const batch = await gitExecFileAsyncBuffer(['cat-file', '--batch'], {
|
||||
...options,
|
||||
stdin: objects.map((read) => `${read.object}\n`).join(''),
|
||||
maxBuffer: objects.reduce((total, read) => total + read.size + 128, 0)
|
||||
})
|
||||
const contents = readPluginGitBatchBlobs(batch.stdout, objects)
|
||||
for (const read of reads) {
|
||||
files.set(read.path, { size: read.size, bytes: contents.get(read.object) ?? null })
|
||||
}
|
||||
} catch {
|
||||
for (const read of reads) {
|
||||
files.set(read.path, null)
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
function chunkByLength(paths: string[], maxChars: number): string[][] {
|
||||
const chunks: string[][] = []
|
||||
let chars = maxChars
|
||||
for (const path of paths) {
|
||||
if (chars + path.length > maxChars) {
|
||||
chunks.push([])
|
||||
chars = 0
|
||||
}
|
||||
chunks.at(-1)!.push(path)
|
||||
chars += path.length + 1
|
||||
}
|
||||
return chunks
|
||||
}
|
||||
|
||||
/** Git for Windows accepts either separator in a pathspec but always lists `/`. */
|
||||
function gitTreePath(path: string): string {
|
||||
return process.platform === 'win32' ? path.replaceAll('\\', '/') : path
|
||||
}
|
||||
|
||||
export function parsePluginMarketplaceIndex(filename: string, bytes: Buffer): PluginMarketplace {
|
||||
|
||||
Reference in New Issue
Block a user