From 604eb6af2d404422b00ca7aec687c7bbff7b6be0 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 14 Jul 2026 04:20:50 +0100 Subject: [PATCH] refactor(concurrency): use native p-limit APIs --- extensions/openshell/src/mirror.ts | 14 ++----- extensions/qa-lab/src/evidence-gallery.ts | 28 ++++++------- src/infra/bonjour-discovery.ts | 51 +++++++++-------------- 3 files changed, 35 insertions(+), 58 deletions(-) diff --git a/extensions/openshell/src/mirror.ts b/extensions/openshell/src/mirror.ts index caf547f4a31e..3af915537d70 100644 --- a/extensions/openshell/src/mirror.ts +++ b/extensions/openshell/src/mirror.ts @@ -35,7 +35,7 @@ async function copyTreeWithoutSymlinks(params: { return; } if (stats.isDirectory()) { - await runLimitedFs(async () => await fs.mkdir(params.targetPath, { recursive: true })); + await runLimitedFs(fs.mkdir, params.targetPath, { recursive: true }); const entries = await runLimitedFs(async () => await fs.readdir(params.sourcePath)); await Promise.all( entries.map(async (entry) => { @@ -49,9 +49,7 @@ async function copyTreeWithoutSymlinks(params: { return; } if (stats.isFile()) { - await runLimitedFs( - async () => await fs.mkdir(path.dirname(params.targetPath), { recursive: true }), - ); + await runLimitedFs(fs.mkdir, path.dirname(params.targetPath), { recursive: true }); await runLimitedFs(async () => await fs.copyFile(params.sourcePath, params.targetPath)); } } @@ -74,13 +72,7 @@ export async function replaceDirectoryContents(params: { if (stats?.isSymbolicLink()) { return; } - await runLimitedFs( - async () => - await fs.rm(targetPath, { - recursive: true, - force: true, - }), - ); + await runLimitedFs(fs.rm, targetPath, { recursive: true, force: true }); }), ); const sourceEntries = await fs.readdir(params.sourceDir); diff --git a/extensions/qa-lab/src/evidence-gallery.ts b/extensions/qa-lab/src/evidence-gallery.ts index 7fe9e28ee8c3..e3bcdc72f293 100644 --- a/extensions/qa-lab/src/evidence-gallery.ts +++ b/extensions/qa-lab/src/evidence-gallery.ts @@ -892,21 +892,19 @@ export async function buildQaEvidenceGalleryModel(params: { repoRoot, }); return { - artifacts: await Promise.all( - (entry.execution?.artifacts ?? []).map((artifact, artifactIndex) => - limitArtifactView(() => - buildArtifactView({ - allowedArtifactFiles, - artifact, - artifactIndex, - evidenceDir, - entryIndex, - extraRoots: [requestedRepoRoot], - hrefEvidencePath, - repoRoot, - }), - ), - ), + artifacts: await limitArtifactView.map( + entry.execution?.artifacts ?? [], + (artifact, artifactIndex) => + buildArtifactView({ + allowedArtifactFiles, + artifact, + artifactIndex, + evidenceDir, + entryIndex, + extraRoots: [requestedRepoRoot], + hrefEvidencePath, + repoRoot, + }), ), coverage: entry.coverage.map((coverage) => ({ id: sanitizeEntryText(coverage.id), diff --git a/src/infra/bonjour-discovery.ts b/src/infra/bonjour-discovery.ts index 4eec464d4fc4..1e04b387a388 100644 --- a/src/infra/bonjour-discovery.ts +++ b/src/infra/bonjour-discovery.ts @@ -5,6 +5,7 @@ import { normalizeStringEntries, uniqueStrings, } from "@openclaw/normalization-core/string-normalization"; +import pLimit from "p-limit"; import { runCommandWithTimeout } from "../process/exec.js"; import { parseStrictInteger } from "./parse-finite-number.js"; import { isTailnetIPv4 } from "./tailnet.js"; @@ -362,45 +363,31 @@ async function discoverWideAreaViaTailnetDns( const probeName = `${GATEWAY_SERVICE_TYPE}.${domain.replace(/\.$/, "")}`; - const concurrency = 6; - let nextIndex = 0; let nameserver: string | null = null; let ptrs: string[] = []; - const worker = async () => { - while (nameserver === null) { - const budget = remainingMs(); - if (budget <= 0) { - return; - } - const i = nextIndex; - nextIndex += 1; - if (i >= ips.length) { - return; - } - const ip = ips[i] ?? ""; - if (!ip) { - continue; - } - try { - const probe = await run( - ["dig", "+short", "+time=1", "+tries=1", `@${ip}`, probeName, "PTR"], - { timeoutMs: Math.max(1, Math.min(250, budget)) }, - ); - const lines = parseDigShortLines(probe.stdout); - if (lines.length === 0) { - continue; - } + await pLimit(6).map(ips, async (ip) => { + if (nameserver !== null) { + return; + } + const budget = remainingMs(); + if (budget <= 0 || !ip) { + return; + } + try { + const probe = await run( + ["dig", "+short", "+time=1", "+tries=1", `@${ip}`, probeName, "PTR"], + { timeoutMs: Math.max(1, Math.min(250, budget)) }, + ); + const lines = parseDigShortLines(probe.stdout); + if (lines.length > 0) { nameserver = ip; ptrs = lines; - return; - } catch { - // ignore } + } catch { + // ignore } - }; - - await Promise.all(Array.from({ length: Math.min(concurrency, ips.length) }, () => worker())); + }); if (!nameserver || ptrs.length === 0) { return [];