From cfa560489ca6fb699b99db36573c411ff83f9df4 Mon Sep 17 00:00:00 2001 From: Zoltan Kochan Date: Thu, 9 Apr 2026 00:21:36 +0200 Subject: [PATCH] perf: send pre-packed msgpack store index entries from server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Instead of serializing file indexes as JSON (server) → parsing (client) → rebuilding Maps → re-encoding as msgpack, the server now sends the raw msgpack buffers from its store index directly. The client writes them straight to SQLite without any encoding. - Add StoreIndex.getRaw() for reading without msgpack decode - Server includes binary msgpack entries in /v1/install response - Client parses binary entries and writes directly via setRawMany - Eliminates ~10MB JSON serialization/parsing round-trip --- registry/client/src/fetchFromPnpmRegistry.ts | 80 +++++++++----------- registry/server/src/createRegistryServer.ts | 47 ++++++++++-- registry/server/src/diff.ts | 42 ++++++---- store/index/src/index.ts | 8 ++ 4 files changed, 114 insertions(+), 63 deletions(-) diff --git a/registry/client/src/fetchFromPnpmRegistry.ts b/registry/client/src/fetchFromPnpmRegistry.ts index 58b7a20029..5e6acfe0d3 100644 --- a/registry/client/src/fetchFromPnpmRegistry.ts +++ b/registry/client/src/fetchFromPnpmRegistry.ts @@ -4,8 +4,7 @@ import { URL } from 'node:url' import { gunzipSync } from 'node:zlib' import type { LockfileObject } from '@pnpm/lockfile.types' -import type { PackageFilesIndex } from '@pnpm/store.cafs' -import { packForStorage, StoreIndex, storeIndexKey } from '@pnpm/store.index' +import { StoreIndex } from '@pnpm/store.index' import { writeCafsFiles } from '@pnpm/worker' import { decodeResponse, type ResponseMetadata } from './protocol.js' @@ -62,17 +61,17 @@ export async function fetchFromPnpmRegistry ( storeIntegrities, }) - // 3. Send resolve request (returns JSON — lockfile + file indexes, no file content) + // 3. Send resolve request — returns binary: JSON metadata + pre-packed msgpack index entries const responseBuffer = await sendRequest(opts.registryUrl, '/v1/install', requestBody) - const metadata: ResponseMetadata = JSON.parse(responseBuffer.toString('utf-8')) + const { metadata, indexEntries } = parseInstallResponse(responseBuffer) // 4. Fetch missing files in parallel batches via /v1/files if (metadata.missingDigests.length > 0) { await fetchFilesInParallel(opts.registryUrl, metadata, opts.storeDir) } - // 5. Write store index entries for new packages - writeStoreIndexEntries(metadata, opts.storeIndex) + // 5. Write pre-packed store index entries directly to SQLite (no msgpack re-encoding) + writeRawIndexEntries(indexEntries, opts.storeIndex) return { lockfile: metadata.lockfile, @@ -207,44 +206,45 @@ async function sendRequest (registryUrl: string, urlPath: string, body: string): }) } -function writeStoreIndexEntries ( - metadata: ResponseMetadata, - storeIndex: StoreIndex -): void { - const writes: Array<{ key: string, buffer: Uint8Array }> = [] +function parseInstallResponse (buf: Buffer): { + metadata: ResponseMetadata + indexEntries: Array<{ key: string, buffer: Uint8Array }> +} { + let offset = 0 - for (const [depPath, pkgFilesInfo] of Object.entries(metadata.packageFiles)) { - // Strip peer suffix from depPath to get pkgId — matches what - // headlessInstall uses via packageIdFromSnapshot() → tryGetPackageId(). - // e.g., "@babel/helper@7.28.6(@babel/core@7.29.0)" → "@babel/helper@7.28.6" - const pkgId = stripPeerSuffix(depPath) - const key = storeIndexKey(pkgFilesInfo.integrity, pkgId) + // Read JSON metadata + const jsonLen = buf.readUInt32BE(offset) + offset += 4 + const metadata: ResponseMetadata = JSON.parse(buf.subarray(offset, offset + jsonLen).toString('utf-8')) + offset += jsonLen - // Check if already in index - if (storeIndex.has(key)) continue + // Read pre-packed msgpack index entries + const indexEntries: Array<{ key: string, buffer: Uint8Array }> = [] + while (offset < buf.length) { + const keyLen = buf.readUInt16BE(offset) + offset += 2 + if (keyLen === 0) break // end marker - // Build PackageFilesIndex - const files = new Map() - for (const [relativePath, fileInfo] of Object.entries(pkgFilesInfo.files)) { - files.set(relativePath, { - checkedAt: Date.now(), - digest: fileInfo.digest, - mode: fileInfo.mode, - size: fileInfo.size, - }) - } + const key = buf.subarray(offset, offset + keyLen).toString('utf-8') + offset += keyLen - const packageFilesIndex: PackageFilesIndex = { - algo: pkgFilesInfo.algo, - files, - } + const bufLen = buf.readUInt32BE(offset) + offset += 4 - writes.push({ - key, - buffer: packForStorage(packageFilesIndex) as Uint8Array, - }) + const buffer = new Uint8Array(buf.buffer, buf.byteOffset + offset, bufLen) + offset += bufLen + + indexEntries.push({ key, buffer }) } + return { metadata, indexEntries } +} + +function writeRawIndexEntries ( + indexEntries: Array<{ key: string, buffer: Uint8Array }>, + storeIndex: StoreIndex +): void { + const writes = indexEntries.filter(({ key }) => !storeIndex.has(key)) if (writes.length > 0) { storeIndex.setRawMany(writes) } @@ -259,12 +259,6 @@ function decompressIfNeeded (buf: Buffer): Buffer { return buf } -function stripPeerSuffix (depPath: string): string { - const parenIdx = depPath.indexOf('(') - if (parenIdx !== -1) return depPath.substring(0, parenIdx) - return depPath -} - async function * toAsyncIterable (buffer: Buffer): AsyncIterable { yield buffer } diff --git a/registry/server/src/createRegistryServer.ts b/registry/server/src/createRegistryServer.ts index 9182deaada..9dfc386cc4 100644 --- a/registry/server/src/createRegistryServer.ts +++ b/registry/server/src/createRegistryServer.ts @@ -8,12 +8,12 @@ import type { InstallOptions } from '@pnpm/installing.deps-installer' import { install } from '@pnpm/installing.deps-installer' import { readWantedLockfile, writeWantedLockfile } from '@pnpm/lockfile.fs' import type { LockfileObject } from '@pnpm/lockfile.types' -import { getFilePathByModeInCafs, type PackageFilesIndex } from '@pnpm/store.cafs' +import { getFilePathByModeInCafs } from '@pnpm/store.cafs' import { createPackageStore, type StoreController } from '@pnpm/store.controller' import { StoreIndex } from '@pnpm/store.index' import type { Registries } from '@pnpm/types' -import { buildIntegrityIndex, computeDiff } from './diff.js' +import { buildIntegrityIndex, computeDiff, type IntegrityEntry } from './diff.js' import { encodeResponse, type MissingFile } from './protocol.js' export interface RegistryServerOptions { @@ -33,7 +33,7 @@ interface ServerContext { storeDir: string cacheDir: string registries: Registries - integrityIndex: Map + integrityIndex: Map } export async function createRegistryServer (opts: RegistryServerOptions): Promise { @@ -213,16 +213,49 @@ async function handleInstall ( ctx.integrityIndex = integrityIndex // Compute the file diff using only what the server already has - const { metadata } = computeDiff( + const { metadata, packageIndexBuffers } = computeDiff( resolvedLockfile, request.storeIntegrities ?? [], integrityIndex, ctx.storeDir ) - // Return JSON only — file contents are fetched via parallel /v1/files requests - res.writeHead(200, { 'Content-Type': 'application/json' }) - res.end(JSON.stringify(metadata)) + // Response format: + // [4 bytes: JSON length][JSON metadata] + // [package index entries: 2B key_len + key + 4B buf_len + msgpack buffer]... + // [2 zero bytes: end marker] + const parts: Buffer[] = [] + const jsonBuf = Buffer.from(JSON.stringify(metadata), 'utf-8') + const jsonLenBuf = Buffer.alloc(4) + jsonLenBuf.writeUInt32BE(jsonBuf.length, 0) + parts.push(jsonLenBuf) + parts.push(jsonBuf) + + for (const [depPath, rawBuffer] of packageIndexBuffers) { + const integrity = metadata.packageFiles[depPath]?.integrity + if (!integrity) continue + // Strip peer suffix to match what the client needs for store index keys + const pkgId = depPath.includes('(') ? depPath.substring(0, depPath.indexOf('(')) : depPath + const key = `${integrity}\t${pkgId}` + const keyBuf = Buffer.from(key, 'utf-8') + const keyLenBuf = Buffer.alloc(2) + keyLenBuf.writeUInt16BE(keyBuf.length, 0) + const bufLenBuf = Buffer.alloc(4) + bufLenBuf.writeUInt32BE(rawBuffer.length, 0) + parts.push(keyLenBuf) + parts.push(keyBuf) + parts.push(bufLenBuf) + parts.push(Buffer.from(rawBuffer)) + } + // End marker + parts.push(Buffer.alloc(2, 0)) + + const payload = Buffer.concat(parts) + res.writeHead(200, { + 'Content-Type': 'application/x-pnpm-install', + 'Content-Length': payload.length, + }) + res.end(payload) } finally { await fs.rm(tmpDir, { recursive: true, force: true }) } diff --git a/registry/server/src/diff.ts b/registry/server/src/diff.ts index 5c511521ac..541350d0f1 100644 --- a/registry/server/src/diff.ts +++ b/registry/server/src/diff.ts @@ -4,17 +4,28 @@ import type { StoreIndex } from '@pnpm/store.index' import type { MissingFile, PackageFilesInfo, ResponseMetadata } from './protocol.js' +export interface IntegrityEntry { + decoded: PackageFilesIndex + rawBuffer: Uint8Array +} + /** - * Build an index mapping integrity hashes to their PackageFilesIndex. - * This is called once at startup and updated as new packages are fetched. + * Build an index mapping integrity hashes to their PackageFilesIndex + * (decoded for diff computation) and raw msgpack buffer (for sending to client). */ -export function buildIntegrityIndex (storeIndex: StoreIndex): Map { - const index = new Map() +export function buildIntegrityIndex (storeIndex: StoreIndex): Map { + const index = new Map() for (const [key, value] of storeIndex.entries()) { const tabIdx = key.indexOf('\t') if (tabIdx === -1) continue const integrity = key.slice(0, tabIdx) - index.set(integrity, value as PackageFilesIndex) + if (index.has(integrity)) continue + const rawBuffer = storeIndex.getRaw(key) + if (!rawBuffer) continue + index.set(integrity, { + decoded: value as PackageFilesIndex, + rawBuffer, + }) } return index } @@ -22,6 +33,8 @@ export function buildIntegrityIndex (storeIndex: StoreIndex): Map } /** @@ -37,7 +50,7 @@ export interface DiffResult { export function computeDiff ( lockfile: LockfileObject, storeIntegrities: string[], - integrityIndex: Map, + integrityIndex: Map, storeDir: string ): DiffResult { // 1. Build the set of file digests the client already has @@ -45,15 +58,16 @@ export function computeDiff ( const clientIntegrities = new Set(storeIntegrities) for (const integrity of storeIntegrities) { - const pkgIndex = integrityIndex.get(integrity) - if (!pkgIndex) continue - for (const [, fileInfo] of getFilesEntries(pkgIndex)) { + const entry = integrityIndex.get(integrity) + if (!entry) continue + for (const [, fileInfo] of getFilesEntries(entry.decoded)) { clientDigests.add(fileInfo.digest) } } // 2. Iterate resolved packages and compute missing files const packageFiles: Record = {} + const packageIndexBuffers = new Map() const missingFiles: MissingFile[] = [] const missingDigests: string[] = [] @@ -76,13 +90,13 @@ export function computeDiff ( continue } - const pkgIndex = integrityIndex.get(integrity) - if (!pkgIndex) continue // package not indexed on server yet + const entry = integrityIndex.get(integrity) + if (!entry) continue // package not indexed on server yet packagesToFetch++ const filesRecord: Record = {} - for (const [relativePath, fileInfo] of getFilesEntries(pkgIndex)) { + for (const [relativePath, fileInfo] of getFilesEntries(entry.decoded)) { filesInNewPackages++ filesRecord[relativePath] = { digest: fileInfo.digest, @@ -108,9 +122,10 @@ export function computeDiff ( packageFiles[depPath] = { integrity, - algo: pkgIndex.algo, + algo: entry.decoded.algo, files: filesRecord, } + packageIndexBuffers.set(depPath, entry.rawBuffer) } return { @@ -129,6 +144,7 @@ export function computeDiff ( }, }, missingFiles, + packageIndexBuffers, } } diff --git a/store/index/src/index.ts b/store/index/src/index.ts index 6dc79d4b56..ba45091bf4 100644 --- a/store/index/src/index.ts +++ b/store/index/src/index.ts @@ -142,6 +142,14 @@ export class StoreIndex { return undefined } + /** + * Get the raw msgpack-encoded buffer for a key without decoding. + */ + getRaw (key: string): Uint8Array | undefined { + const row = sqliteRetry(() => this.stmtGet.get(key)) as { data: Uint8Array } | undefined + return row?.data + } + set (key: string, data: unknown): void { const buffer = packr.pack(data) sqliteRetry(() => {