perf(scan): 后台扫描改为 fs.promises + 并发池并行 IO
scanStorageAsync 重写：异步递归列目录 + 16 路并发池并行 扫描包目录，目录内文件 stat 并行，结果按下标回填保持顺序。 数千包/慢磁盘（NAS）场景吞吐显著优于旧的单线程同步 IO。
This commit is contained in:
陈银军
+124
-72
@@ -245,114 +245,166 @@ export function scanStorage(storageDir) {
|
||||
return packages
|
||||
}
|
||||
|
||||
/** 异步递归列文件（fs.promises，目录并行读取） */
|
||||
async function listFilesAsync(dir, depth = Infinity) {
|
||||
const out = []
|
||||
async function walk(d, dth) {
|
||||
let entries
|
||||
try {
|
||||
entries = await fs.promises.readdir(d, { withFileTypes: true })
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
await Promise.all(entries.map(async e => {
|
||||
if (e.name === 'node_modules' || e.name.startsWith('.') || e.name === '.DS_Store') return
|
||||
const full = path.join(d, e.name)
|
||||
if (e.isDirectory()) {
|
||||
if (dth > 0) await walk(full, dth - 1)
|
||||
} else out.push(full)
|
||||
}))
|
||||
}
|
||||
await walk(dir, depth)
|
||||
return out
|
||||
}
|
||||
|
||||
/**
|
||||
* 异步扫描（分片让出事件循环 + 进度回调）
|
||||
* 用于后台重扫：扫描期间仍能响应 /api/scan/progress 等请求，前端可实时展示进度
|
||||
* 异步扫描（fs.promises + 并发池，多目录并行 IO）
|
||||
* 用于后台重扫：并发扫描期间仍能响应 /api/scan/progress 等请求，前端可实时展示进度
|
||||
* 相比旧版（逐个目录同步 IO + setImmediate 让出），数千包场景吞吐明显提升
|
||||
*/
|
||||
export async function scanStorageAsync(storageDir, onProgress) {
|
||||
const packages = []
|
||||
if (!isValidStorage(storageDir)) return packages
|
||||
|
||||
const allFiles = listFiles(storageDir)
|
||||
const allFiles = await listFilesAsync(storageDir)
|
||||
// 包目录 = 含 package.json 的目录 ∪ 含 .tgz 的目录（避免只缓存了 tgz、缺元数据的目录被漏算）
|
||||
const pkgDirs = new Set()
|
||||
for (const f of allFiles) {
|
||||
if (f.endsWith('package.json') || f.endsWith('.tgz')) pkgDirs.add(path.dirname(f))
|
||||
}
|
||||
const storageRoot = path.resolve(storageDir)
|
||||
const dirs = [...pkgDirs].sort() // 排序保证输出顺序稳定（异步递归收集顺序不固定）
|
||||
let scanned = 0
|
||||
let scannedBytes = 0 // 所有文件字节和（总占用，与 du 同口径）
|
||||
let tgzBytes = 0 // 仅 .tgz 字节和（包粒度用）
|
||||
|
||||
for (const pkgDir of pkgDirs) {
|
||||
// 每处理完一个包目录就让出事件循环，保证进度接口可实时响应
|
||||
await new Promise(r => setImmediate(r))
|
||||
// 单包目录扫描：目录内文件 stat 并行，元数据损坏时按物理 .tgz 计算真实占用
|
||||
async function scanOne(pkgDir) {
|
||||
const pkgFile = path.join(pkgDir, 'package.json')
|
||||
|
||||
const tgzSizes = new Map()
|
||||
let tgzTotal = 0
|
||||
let tgzLatest = 0 // 最新 .tgz 文件 mtime（毫秒）：今天缓存了新版本也算「今天更新」
|
||||
let entries
|
||||
try {
|
||||
for (const f of fs.readdirSync(pkgDir)) {
|
||||
const st = fs.statSync(path.join(pkgDir, f))
|
||||
const size = st.size
|
||||
scannedBytes += size
|
||||
if (f.endsWith('.tgz')) {
|
||||
tgzTotal += size
|
||||
tgzSizes.set(f.slice(0, -4), size)
|
||||
if (st.mtimeMs > tgzLatest) tgzLatest = st.mtimeMs
|
||||
}
|
||||
entries = await fs.promises.readdir(pkgDir, { withFileTypes: true })
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
const stats = (await Promise.all(entries.map(async e => {
|
||||
try {
|
||||
return { name: e.name, st: await fs.promises.stat(path.join(pkgDir, e.name)) }
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
} catch {}
|
||||
}))).filter(Boolean)
|
||||
for (const { name, st } of stats) {
|
||||
scannedBytes += st.size
|
||||
if (name.endsWith('.tgz')) {
|
||||
tgzTotal += st.size
|
||||
tgzSizes.set(name.slice(0, -4), st.size)
|
||||
if (st.mtimeMs > tgzLatest) tgzLatest = st.mtimeMs
|
||||
}
|
||||
}
|
||||
tgzBytes += tgzTotal
|
||||
|
||||
const meta = readJson(pkgFile)
|
||||
let meta = null
|
||||
try {
|
||||
meta = JSON.parse(await fs.promises.readFile(pkgFile, 'utf8'))
|
||||
} catch {}
|
||||
if (!meta) {
|
||||
packages.push({
|
||||
return {
|
||||
name: pkgNameFromDir(pkgDir, storageRoot),
|
||||
status: 'broken',
|
||||
versions: 0,
|
||||
size: fmtMb(tgzTotal),
|
||||
lastUpdated: latestDate(null, tgzLatest),
|
||||
versionsDetail: [],
|
||||
})
|
||||
} else {
|
||||
const versionsObj = meta.versions || {}
|
||||
const timeObj = meta.time || {}
|
||||
const distTags = meta['dist-tags'] || {}
|
||||
// 包信息（来自本地元数据，离线可用）：介绍/主页/仓库/作者/许可
|
||||
// 顶层缺失时回退到最新版本里的字段（Verdaccio 落盘元数据顶层不含这些）
|
||||
const description = metaField(meta, versionsObj, 'description')
|
||||
const homepage = metaField(meta, versionsObj, 'homepage')
|
||||
const repository = metaField(meta, versionsObj, 'repository', r => r.url)
|
||||
const author = metaField(meta, versionsObj, 'author', a => a.name)
|
||||
const license = metaField(meta, versionsObj, 'license')
|
||||
const versionsDetail = []
|
||||
const totalBytes = tgzTotal
|
||||
for (const [v, info] of Object.entries(versionsObj)) {
|
||||
let size = tgzSizes.get(v)
|
||||
if (typeof size !== 'number') {
|
||||
const tb = info && info.dist && info.dist.tarball
|
||||
const bn = tb ? String(tb).split('/').pop() : ''
|
||||
if (bn.endsWith('.tgz')) {
|
||||
const hit = tgzSizes.get(bn.slice(0, -4))
|
||||
if (typeof hit === 'number') size = hit
|
||||
}
|
||||
}
|
||||
// 无物理 .tgz 文件的版本是元数据残留（实体已被清理），不计数，避免版本数虚高
|
||||
if (typeof size !== 'number') continue
|
||||
versionsDetail.push({
|
||||
version: parseVersion(v),
|
||||
size: fmtMb(size),
|
||||
date: timeObj[v] ? shDate(timeObj[v]) : null,
|
||||
})
|
||||
}
|
||||
versionsDetail.sort((a, b) => a.version.localeCompare(b.version, undefined, { numeric: true }))
|
||||
// 最近更新：取「元数据最新版本发布日」与「最新 .tgz 缓存日」中较新者（今天拉缓存也算今天更新）
|
||||
const metaUpdated = Object.keys(timeObj).length
|
||||
? Object.entries(timeObj).map(([, iso]) => shDate(iso)).filter(Boolean).sort().pop()
|
||||
: null
|
||||
const lastUpdated = latestDate(metaUpdated, tgzLatest)
|
||||
packages.push({
|
||||
name: pkgNameFromDir(pkgDir, storageRoot),
|
||||
versions: versionsDetail.length,
|
||||
size: fmtMb(totalBytes),
|
||||
lastUpdated,
|
||||
visits: 0,
|
||||
distTags,
|
||||
description,
|
||||
homepage,
|
||||
repository,
|
||||
author,
|
||||
license,
|
||||
versionsDetail,
|
||||
})
|
||||
}
|
||||
|
||||
scanned += 1
|
||||
if (onProgress) onProgress({ packages: scanned, bytes: scannedBytes, tgzBytes })
|
||||
const versionsObj = meta.versions || {}
|
||||
const timeObj = meta.time || {}
|
||||
const distTags = meta['dist-tags'] || {}
|
||||
// 包信息（来自本地元数据，离线可用）：介绍/主页/仓库/作者/许可
|
||||
// 顶层缺失时回退到最新版本里的字段（Verdaccio 落盘元数据顶层不含这些）
|
||||
const description = metaField(meta, versionsObj, 'description')
|
||||
const homepage = metaField(meta, versionsObj, 'homepage')
|
||||
const repository = metaField(meta, versionsObj, 'repository', r => r.url)
|
||||
const author = metaField(meta, versionsObj, 'author', a => a.name)
|
||||
const license = metaField(meta, versionsObj, 'license')
|
||||
const versionsDetail = []
|
||||
// 总大小以实际 .tgz 文件为准
|
||||
const totalBytes = tgzTotal
|
||||
for (const [v, info] of Object.entries(versionsObj)) {
|
||||
let size = tgzSizes.get(v)
|
||||
if (typeof size !== 'number') {
|
||||
// 兼容 {pkgname}-{version}.tgz 命名：用元数据 dist.tarball 的文件名精确匹配
|
||||
const tb = info && info.dist && info.dist.tarball
|
||||
const bn = tb ? String(tb).split('/').pop() : ''
|
||||
if (bn.endsWith('.tgz')) {
|
||||
const hit = tgzSizes.get(bn.slice(0, -4))
|
||||
if (typeof hit === 'number') size = hit
|
||||
}
|
||||
}
|
||||
// 无物理 .tgz 文件的版本是元数据残留（实体已被清理），不计数，避免版本数虚高
|
||||
if (typeof size !== 'number') continue
|
||||
versionsDetail.push({
|
||||
version: parseVersion(v),
|
||||
size: fmtMb(size),
|
||||
date: timeObj[v] ? shDate(timeObj[v]) : null,
|
||||
})
|
||||
}
|
||||
// 按 semver 数值排序（旧 → 新）
|
||||
versionsDetail.sort((a, b) => a.version.localeCompare(b.version, undefined, { numeric: true }))
|
||||
|
||||
// 最近更新：取「元数据最新版本发布日」与「最新 .tgz 缓存日」中较新者（今天拉缓存也算今天更新）
|
||||
const metaUpdated = Object.keys(timeObj).length
|
||||
? Object.entries(timeObj).map(([, iso]) => shDate(iso)).filter(Boolean).sort().pop()
|
||||
: null
|
||||
const lastUpdated = latestDate(metaUpdated, tgzLatest)
|
||||
|
||||
return {
|
||||
name: pkgNameFromDir(pkgDir, storageRoot),
|
||||
versions: versionsDetail.length,
|
||||
size: fmtMb(totalBytes),
|
||||
lastUpdated,
|
||||
visits: 0, // 真实环境无访问统计，界面不依赖真实值
|
||||
distTags,
|
||||
description,
|
||||
homepage,
|
||||
repository,
|
||||
author,
|
||||
license,
|
||||
versionsDetail,
|
||||
}
|
||||
}
|
||||
|
||||
// 并发池：同时处理 CONCURRENCY 个包目录，结果按下标回填保持稳定顺序
|
||||
const CONCURRENCY = 16
|
||||
const results = new Array(dirs.length)
|
||||
let next = 0
|
||||
const worker = async () => {
|
||||
while (true) {
|
||||
const idx = next++
|
||||
if (idx >= dirs.length) return
|
||||
results[idx] = await scanOne(dirs[idx])
|
||||
scanned += 1
|
||||
if (onProgress) onProgress({ packages: scanned, bytes: scannedBytes, tgzBytes })
|
||||
}
|
||||
}
|
||||
await Promise.all(Array.from({ length: Math.min(CONCURRENCY, dirs.length) }, worker))
|
||||
|
||||
for (const p of results) if (p) packages.push(p)
|
||||
return packages
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user