From 8ffbcf05595758187e27e3d4038317f5e82dd29b Mon Sep 17 00:00:00 2001 From: darshanime Date: Tue, 14 Jul 2026 17:37:59 +0530 Subject: [PATCH] add o11y for multiget --- dist/index.js | 29 +++++++++++++++++++------- src/warpbuild/mirror-cache.ts | 38 ++++++++++++++++++++++++++++------- 2 files changed, 53 insertions(+), 14 deletions(-) diff --git a/dist/index.js b/dist/index.js index a0d853c..c4d1a7f 100644 --- a/dist/index.js +++ b/dist/index.js @@ -41990,8 +41990,9 @@ async function contribute(settings) { // nothing is extracted into .git directly. async function seedBundle(repoPath, url, refNs) { const tmp = tempBundlePath('seed'); + const what = refNs === BASE_REFNS ? 'base' : 'branch'; try { - await downloadTo(url, tmp); + await downloadTo(url, tmp, what); await exec_exec('git', [ '-C', repoPath, @@ -42150,18 +42151,22 @@ async function hasBaseRefs(repoPath) { // Download a presigned GET to dest with concurrent HTTP range requests, so the base bundle // (tens/hundreds of MiB) transfers at link rate instead of the single-stream ~15 MB/s ceiling. // Falls back to a single stream when the server doesn't support ranges. -async function downloadTo(url, dest) { +async function downloadTo(url, dest, what) { + const started = Date.now(); const total = await probeRangeSize(url); if (total === null) { await downloadSingleStream(url, dest); + const bytes = (await external_fs_namespaceObject.promises.stat(dest)).size; + info(`Mirror download [${what}]: ${fmtSize(bytes)} single-stream, no range support, ${fmtElapsed(started, bytes)}`); return; } + const segments = []; + for (let off = 0; off < total; off += SEGMENT_SIZE) { + segments.push([off, Math.min(SEGMENT_SIZE, total - off)]); + } + const width = Math.min(SEGMENT_CONCURRENCY, segments.length); const fh = await external_fs_namespaceObject.promises.open(dest, 'w'); try { - const segments = []; - for (let off = 0; off < total; off += SEGMENT_SIZE) { - segments.push([off, Math.min(SEGMENT_SIZE, total - off)]); - } let next = 0; const worker = async () => { for (;;) { @@ -42175,7 +42180,7 @@ async function downloadTo(url, dest) { } }; const pool = []; - for (let k = 0; k < Math.min(SEGMENT_CONCURRENCY, segments.length); k++) { + for (let k = 0; k < width; k++) { pool.push(worker()); } await Promise.all(pool); @@ -42183,6 +42188,16 @@ async function downloadTo(url, dest) { finally { await fh.close(); } + info(`Mirror download [${what}]: ${fmtSize(total)} in ${segments.length} ranged segments ×${width}, ${fmtElapsed(started, total)}`); +} +// Size + elapsed/throughput for the mirror download o11y line. +function fmtSize(bytes) { + return `${(bytes / 1e6).toFixed(1)} MB`; +} +function fmtElapsed(startedMs, bytes) { + const secs = (Date.now() - startedMs) / 1000; + const rate = secs > 0 ? (bytes / 1e6 / secs).toFixed(1) : 'n/a'; + return `${secs.toFixed(2)}s (${rate} MB/s)`; } // GET bytes=0-0 to learn the total size and confirm range support; null → not supported. async function probeRangeSize(url) { diff --git a/src/warpbuild/mirror-cache.ts b/src/warpbuild/mirror-cache.ts index 28ffdcc..c2bbfac 100644 --- a/src/warpbuild/mirror-cache.ts +++ b/src/warpbuild/mirror-cache.ts @@ -257,8 +257,9 @@ async function seedBundle( refNs: string ): Promise { const tmp = tempBundlePath('seed') + const what = refNs === BASE_REFNS ? 'base' : 'branch' try { - await downloadTo(url, tmp) + await downloadTo(url, tmp, what) await exec.exec('git', [ '-C', repoPath, @@ -442,18 +443,28 @@ async function hasBaseRefs(repoPath: string): Promise { // Download a presigned GET to dest with concurrent HTTP range requests, so the base bundle // (tens/hundreds of MiB) transfers at link rate instead of the single-stream ~15 MB/s ceiling. // Falls back to a single stream when the server doesn't support ranges. -async function downloadTo(url: string, dest: string): Promise { +async function downloadTo( + url: string, + dest: string, + what: string +): Promise { + const started = Date.now() const total = await probeRangeSize(url) if (total === null) { await downloadSingleStream(url, dest) + const bytes = (await fs.promises.stat(dest)).size + core.info( + `Mirror download [${what}]: ${fmtSize(bytes)} single-stream, no range support, ${fmtElapsed(started, bytes)}` + ) return } + const segments: [number, number][] = [] + for (let off = 0; off < total; off += SEGMENT_SIZE) { + segments.push([off, Math.min(SEGMENT_SIZE, total - off)]) + } + const width = Math.min(SEGMENT_CONCURRENCY, segments.length) const fh = await fs.promises.open(dest, 'w') try { - const segments: [number, number][] = [] - for (let off = 0; off < total; off += SEGMENT_SIZE) { - segments.push([off, Math.min(SEGMENT_SIZE, total - off)]) - } let next = 0 const worker = async (): Promise => { for (;;) { @@ -467,13 +478,26 @@ async function downloadTo(url: string, dest: string): Promise { } } const pool: Promise[] = [] - for (let k = 0; k < Math.min(SEGMENT_CONCURRENCY, segments.length); k++) { + for (let k = 0; k < width; k++) { pool.push(worker()) } await Promise.all(pool) } finally { await fh.close() } + core.info( + `Mirror download [${what}]: ${fmtSize(total)} in ${segments.length} ranged segments ×${width}, ${fmtElapsed(started, total)}` + ) +} + +// Size + elapsed/throughput for the mirror download o11y line. +function fmtSize(bytes: number): string { + return `${(bytes / 1e6).toFixed(1)} MB` +} +function fmtElapsed(startedMs: number, bytes: number): string { + const secs = (Date.now() - startedMs) / 1000 + const rate = secs > 0 ? (bytes / 1e6 / secs).toFixed(1) : 'n/a' + return `${secs.toFixed(2)}s (${rate} MB/s)` } // GET bytes=0-0 to learn the total size and confirm range support; null → not supported.