diff --git a/extensions/memory-core/package.json b/extensions/memory-core/package.json index b35bbd215ad7..c2ef49fd8f19 100644 --- a/extensions/memory-core/package.json +++ b/extensions/memory-core/package.json @@ -7,6 +7,7 @@ "dependencies": { "chokidar": "5.0.0", "json5": "2.2.3", + "p-limit": "7.3.0", "typebox": "1.3.3" }, "devDependencies": { diff --git a/extensions/memory-core/src/dreaming-narrative.ts b/extensions/memory-core/src/dreaming-narrative.ts index 32bd15683b42..ee9f5a4ae844 100644 --- a/extensions/memory-core/src/dreaming-narrative.ts +++ b/extensions/memory-core/src/dreaming-narrative.ts @@ -16,6 +16,7 @@ import { resolveStateDir } from "openclaw/plugin-sdk/memory-core-host-runtime-co import { getRuntimeConfig } from "openclaw/plugin-sdk/runtime-config-snapshot"; import { cleanupSessionLifecycleArtifacts } from "openclaw/plugin-sdk/session-store-runtime"; import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; +import pLimit from "p-limit"; import { readDreamsFile, resolveDreamsPath, updateDreamsFile } from "./dreaming-dreams-file.js"; // ── Types ────────────────────────────────────────────────────────────── @@ -997,44 +998,20 @@ export async function generateAndAppendDreamNarrative(params: { // write-lock while it runs and burns a model slot, which caused lock // contention (>30 s) and cascading narrative timeouts (#73198). // -// `runDetachedDreamNarrative` wraps `generateAndAppendDreamNarrative` with a -// FIFO queue capped at `DETACHED_NARRATIVE_CONCURRENCY` so the total in-flight -// detached narratives across phases/workspaces stays bounded. +// `runDetachedDreamNarrative` caps total in-flight detached narratives across +// phases/workspaces so cron sweeps cannot exhaust model and session-lock slots. const DETACHED_NARRATIVE_CONCURRENCY = 3; - -let activeDetachedNarratives = 0; -const detachedNarrativeQueue: Array<() => void> = []; - -function releaseDetachedNarrativeSlot(): void { - activeDetachedNarratives -= 1; - detachedNarrativeQueue.shift()?.(); -} - -async function acquireDetachedNarrativeSlot(): Promise { - if (activeDetachedNarratives >= DETACHED_NARRATIVE_CONCURRENCY) { - await new Promise((resolve) => { - detachedNarrativeQueue.push(resolve); - }); - } - activeDetachedNarratives += 1; -} +const detachedNarrativeLimit = pLimit(DETACHED_NARRATIVE_CONCURRENCY); export function runDetachedDreamNarrative( params: Parameters[0], ): void { queueMicrotask(() => { - void (async () => { - await acquireDetachedNarrativeSlot(); - try { - await generateAndAppendDreamNarrative(params); - } catch { - // Detached narratives intentionally swallow errors — callers (cron - // sweeps) cannot recover, and surfacing here would only cause noisy - // unhandled rejections. Logging happens inside - // generateAndAppendDreamNarrative. - } finally { - releaseDetachedNarrativeSlot(); - } - })(); + void detachedNarrativeLimit(() => generateAndAppendDreamNarrative(params)).catch(() => { + // Detached narratives intentionally swallow errors — callers (cron + // sweeps) cannot recover, and surfacing here would only cause noisy + // unhandled rejections. Logging happens inside + // generateAndAppendDreamNarrative. + }); }); } diff --git a/extensions/memory-core/src/short-term-promotion.ts b/extensions/memory-core/src/short-term-promotion.ts index 78b165b13509..786a6028cdc0 100644 --- a/extensions/memory-core/src/short-term-promotion.ts +++ b/extensions/memory-core/src/short-term-promotion.ts @@ -18,6 +18,7 @@ import { uniqueStrings, } from "openclaw/plugin-sdk/string-coerce-runtime"; import { truncateUtf16Safe } from "openclaw/plugin-sdk/text-utility-runtime"; +import pLimit from "p-limit"; import { deriveConceptTags, MAX_CONCEPT_TAGS, @@ -56,6 +57,8 @@ const PROMOTED_SNIPPET_CHARS_PER_TOKEN_ESTIMATE = 4; const MAX_QUERY_HASHES = 32; const MAX_RECALL_DAYS = 16; const SHORT_TERM_RECALL_MAX_ENTRIES = 512; +// One recall batch can inspect every retained entry; cap filesystem pressure. +const SHORT_TERM_SOURCE_FILE_CHECK_CONCURRENCY = 32; const SHORT_TERM_RECALL_MAX_SNIPPET_CHARS = 800; export const SHORT_TERM_STORE_RELATIVE_PATH = path.join( "memory", @@ -1333,12 +1336,13 @@ export async function filterLiveShortTermRecallEntries(params: { return []; } const sourceFileChecks = new Map>(); + const sourceFileLimit = pLimit(SHORT_TERM_SOURCE_FILE_CHECK_CONCURRENCY); const checkSourceFile = (sourcePath: string): Promise => { const existing = sourceFileChecks.get(sourcePath); if (existing) { return existing; } - const check = shortTermRecallSourceIsFile(sourcePath); + const check = sourceFileLimit(() => shortTermRecallSourceIsFile(sourcePath)); sourceFileChecks.set(sourcePath, check); return check; }; diff --git a/extensions/memory-wiki/package.json b/extensions/memory-wiki/package.json index 3fdd55eed517..e659b8d49268 100644 --- a/extensions/memory-wiki/package.json +++ b/extensions/memory-wiki/package.json @@ -6,6 +6,7 @@ "type": "module", "dependencies": { "mdast-util-from-markdown": "2.0.3", + "p-map": "7.0.5", "typebox": "1.3.3", "yaml": "2.9.0", "zod": "4.4.3" diff --git a/extensions/memory-wiki/src/import-runs-state.ts b/extensions/memory-wiki/src/import-runs-state.ts index 860923751ad1..0695744508dd 100644 --- a/extensions/memory-wiki/src/import-runs-state.ts +++ b/extensions/memory-wiki/src/import-runs-state.ts @@ -6,6 +6,9 @@ import type { OpenKeyedStoreOptions, PluginStateKeyedStore, } from "openclaw/plugin-sdk/plugin-state-runtime"; +import pMap from "p-map"; + +const LEGACY_IMPORT_RUN_READ_CONCURRENCY = 16; type ChatGptImportRunEntry = { path: string; @@ -471,13 +474,13 @@ export async function readLegacyMemoryWikiImportRunRecords( } throw error; }); - const records = await Promise.all( - entries - .filter((entry) => entry.isFile() && entry.name.endsWith(".json")) - .map(async (entry) => { - const raw = await fs.readFile(path.join(importRunsDir, entry.name), "utf8"); - return normalizeMemoryWikiImportRunRecord(JSON.parse(raw) as unknown); - }), + const records = await pMap( + entries.filter((entry) => entry.isFile() && entry.name.endsWith(".json")), + async (entry) => { + const raw = await fs.readFile(path.join(importRunsDir, entry.name), "utf8"); + return normalizeMemoryWikiImportRunRecord(JSON.parse(raw) as unknown); + }, + { concurrency: LEGACY_IMPORT_RUN_READ_CONCURRENCY, stopOnError: true }, ); return records.filter((entry): entry is ChatGptImportRunRecord => entry !== null); } diff --git a/extensions/memory-wiki/src/query.ts b/extensions/memory-wiki/src/query.ts index a8931e712ae9..0e2c76b3d8c7 100644 --- a/extensions/memory-wiki/src/query.ts +++ b/extensions/memory-wiki/src/query.ts @@ -18,6 +18,7 @@ import { normalizeLowercaseStringOrEmpty, uniqueStrings, } from "openclaw/plugin-sdk/string-coerce-runtime"; +import pMap from "p-map"; import type { OpenClawConfig } from "../api.js"; import { assessClaimFreshness, isClaimContestedStatus } from "./claim-health.js"; import type { ResolvedMemoryWikiConfig, WikiSearchBackend, WikiSearchCorpus } from "./config.js"; @@ -33,6 +34,7 @@ import { initializeMemoryWikiVault } from "./vault.js"; const QUERY_DIRS = ["entities", "concepts", "sources", "syntheses", "reports"] as const; const AGENT_DIGEST_PATH = ".openclaw-wiki/cache/agent-digest.json"; const CLAIMS_DIGEST_PATH = ".openclaw-wiki/cache/claims.jsonl"; +const QUERY_PAGE_READ_CONCURRENCY = 16; const RELATED_BLOCK_PATTERN = /[\s\S]*?/g; const MARKDOWN_FRONTMATTER_PATTERN = /^\s*---\r?\n[\s\S]*?\r?\n---\r?\n?/; @@ -271,13 +273,15 @@ async function readQueryableWikiPagesByPaths( rootDir: string, files: string[], ): Promise { - const pages = await Promise.all( - files.map(async (relativePath) => { + const pages = await pMap( + files, + async (relativePath) => { const absolutePath = path.join(rootDir, relativePath); const raw = await fs.readFile(absolutePath, "utf8"); const summary = toWikiPageSummary({ absolutePath, relativePath, raw }); return summary ? { ...summary, raw } : null; - }), + }, + { concurrency: QUERY_PAGE_READ_CONCURRENCY, stopOnError: true }, ); return pages.flatMap((page) => (page ? [page] : [])); } diff --git a/extensions/memory-wiki/src/unsafe-local.ts b/extensions/memory-wiki/src/unsafe-local.ts index 8e72f5898f54..0491934a1e40 100644 --- a/extensions/memory-wiki/src/unsafe-local.ts +++ b/extensions/memory-wiki/src/unsafe-local.ts @@ -3,6 +3,7 @@ import { createHash } from "node:crypto"; import fs from "node:fs/promises"; import path from "node:path"; import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime"; +import pMap from "p-map"; import type { BridgeMemoryWikiResult } from "./bridge.js"; import type { ResolvedMemoryWikiConfig } from "./config.js"; import { appendMemoryWikiLog } from "./log.js"; @@ -30,6 +31,7 @@ type UnsafeLocalArtifact = { }; const DIRECTORY_TEXT_EXTENSIONS = new Set([".json", ".jsonl", ".md", ".txt", ".yaml", ".yml"]); +const UNSAFE_LOCAL_SYNC_CONCURRENCY = 16; function detectFenceLanguage(filePath: string): string { const ext = normalizeLowercaseStringOrEmpty(path.extname(filePath)); @@ -222,8 +224,9 @@ export async function syncMemoryWikiUnsafeLocalSources( incomingCount: artifacts.length, }); const activeKeys = new Set(); - const results = await Promise.all( - artifacts.map(async (artifact) => { + const results = await pMap( + artifacts, + async (artifact) => { const stats = await fs.stat(artifact.absolutePath); activeKeys.add(artifact.syncKey); return await writeUnsafeLocalSourcePage({ @@ -233,7 +236,8 @@ export async function syncMemoryWikiUnsafeLocalSources( sourceSize: stats.size, state, }); - }), + }, + { concurrency: UNSAFE_LOCAL_SYNC_CONCURRENCY, stopOnError: true }, ); const removedCount = await pruneImportedSourceEntries({ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index d0865ef46d64..0d0f78ad7b57 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1093,6 +1093,9 @@ importers: json5: specifier: 2.2.3 version: 2.2.3 + p-limit: + specifier: 7.3.0 + version: 7.3.0 typebox: specifier: 1.3.3 version: 1.3.3 @@ -1128,6 +1131,9 @@ importers: mdast-util-from-markdown: specifier: 2.0.3 version: 2.0.3 + p-map: + specifier: 7.0.5 + version: 7.0.5 typebox: specifier: 1.3.3 version: 1.3.3 diff --git a/scripts/run-additional-boundary-checks.mjs b/scripts/run-additional-boundary-checks.mjs index 177cdd869715..9298c8922898 100644 --- a/scripts/run-additional-boundary-checks.mjs +++ b/scripts/run-additional-boundary-checks.mjs @@ -3,6 +3,7 @@ // timeout handling, and grouped CI output. import { spawn } from "node:child_process"; import { performance } from "node:perf_hooks"; +import pMap from "p-map"; import prettyMilliseconds from "pretty-ms"; const DEFAULT_CHECK_TIMEOUT_MS = 10 * 60 * 1000; @@ -499,43 +500,23 @@ export async function runChecks( outputMaxBytes = DEFAULT_OUTPUT_MAX_BYTES, } = {}, ) { - const results = Array.from({ length: checks.length }); const activeChildren = new Set(); const removeActiveChildCleanup = installActiveChildCleanup(activeChildren); - let nextIndex = 0; - let active = 0; + let results; try { - await new Promise((resolve) => { - const launch = () => { - if (nextIndex >= checks.length && active === 0) { - resolve(); - return; - } - - while (active < concurrency && nextIndex < checks.length) { - const index = nextIndex; - const check = checks[nextIndex++]; - active += 1; - void runSingleCheck(check, { - activeChildren, - checkTimeoutMs, - cwd, - env, - outputMaxBytes, - }) - .then((result) => { - results[index] = result; - }) - .finally(() => { - active -= 1; - launch(); - }); - } - }; - - launch(); - }); + results = await pMap( + checks, + (check) => + runSingleCheck(check, { + activeChildren, + checkTimeoutMs, + cwd, + env, + outputMaxBytes, + }), + { concurrency, stopOnError: true }, + ); } finally { removeActiveChildCleanup(); } diff --git a/scripts/run-oxlint-shards.mjs b/scripts/run-oxlint-shards.mjs index 2730bcf5286e..4ded2b7c510f 100644 --- a/scripts/run-oxlint-shards.mjs +++ b/scripts/run-oxlint-shards.mjs @@ -3,6 +3,7 @@ import { spawn, spawnSync } from "node:child_process"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; +import pMap from "p-map"; import { acquireLocalHeavyCheckLockSync, resolveLocalHeavyCheckEnv, @@ -272,21 +273,13 @@ export async function main(extraArgs = process.argv.slice(2), runtimeEnv = proce platform: process.platform, splitCore: shardArgs.splitCore, }); - const results = - shardConcurrency <= 1 - ? await runShardsSerial({ - entries: selectedShards, - env, - extraArgs: shardArgs.oxlintArgs, - runner, - }) - : await runShardsParallel({ - concurrency: Math.min(shardConcurrency, selectedShards.length), - entries: selectedShards, - env, - extraArgs: shardArgs.oxlintArgs, - runner, - }); + const results = await runShards({ + concurrency: Math.max(1, Math.min(shardConcurrency, selectedShards.length)), + entries: selectedShards, + env, + extraArgs: shardArgs.oxlintArgs, + runner, + }); process.exitCode = results.find((status) => status !== 0) ?? 0; } } finally { @@ -385,38 +378,17 @@ export function resolveOxlintShardConcurrency({ ); } -async function runShardsSerial({ entries, env, extraArgs, runner }) { - const results = []; - for (const shard of entries) { - results.push(await runShard({ env, extraArgs, runner, shard })); - if (isParentTerminationRequested()) { - break; - } - } - return results; -} - -async function runShardsParallel({ concurrency, entries, env, extraArgs, runner }) { - const results = []; - results.length = entries.length; - let nextIndex = 0; - - const workers = Array.from({ length: concurrency }, async () => { - for (;;) { +async function runShards({ concurrency, entries, env, extraArgs, runner }) { + const results = await pMap( + entries, + async (shard) => { if (isParentTerminationRequested()) { - return; + return undefined; } - const currentIndex = nextIndex; - nextIndex += 1; - const shard = entries[currentIndex]; - if (!shard) { - return; - } - results[currentIndex] = await runShard({ env, extraArgs, runner, shard }); - } - }); - - await Promise.all(workers); + return await runShard({ env, extraArgs, runner, shard }); + }, + { concurrency, stopOnError: true }, + ); return results.filter((status) => status !== undefined); } diff --git a/scripts/test-extension-batch.mjs b/scripts/test-extension-batch.mjs index 0848d3c8abfd..2095bfa87d5b 100644 --- a/scripts/test-extension-batch.mjs +++ b/scripts/test-extension-batch.mjs @@ -2,6 +2,7 @@ // Runs grouped Vitest plans for one or more bundled plugins. import path from "node:path"; +import pMap from "p-map"; import { listTrackedTestFilesForRoots, resolveExtensionBatchPlan, @@ -196,14 +197,11 @@ export async function runExtensionBatchPlan(batchPlan, params = {}) { console.log(`[test-extension-batch] Running up to ${parallelism} config groups in parallel`); } - let nextGroupIndex = 0; let exitCode = 0; - async function worker() { - while (exitCode === 0) { - const groupIndex = nextGroupIndex; - nextGroupIndex += 1; - const group = orderedGroups[groupIndex]; - if (!group) { + await pMap( + orderedGroups, + async (group, groupIndex) => { + if (exitCode !== 0) { return; } const groupExitCode = await runPlanGroup(group, { @@ -215,14 +213,12 @@ export async function runExtensionBatchPlan(batchPlan, params = {}) { useDedicatedCache, vitestArgs, }); - if (groupExitCode !== 0) { + if (groupExitCode !== 0 && exitCode === 0) { exitCode = groupExitCode; - return; } - } - } - - await Promise.all(Array.from({ length: parallelism }, () => worker())); + }, + { concurrency: parallelism, stopOnError: true }, + ); return exitCode; } diff --git a/scripts/test-group-report.mjs b/scripts/test-group-report.mjs index 679c8615873e..70aaeee99a9e 100644 --- a/scripts/test-group-report.mjs +++ b/scripts/test-group-report.mjs @@ -4,6 +4,7 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; import { pathToFileURL } from "node:url"; +import pMap from "p-map"; import { parsePositiveInt } from "./lib/numeric-options.mjs"; import { buildGroupedTestComparison, @@ -980,19 +981,15 @@ export async function runReportPlans(params) { const concurrency = resolveRunPlanConcurrency(params.args, params.runPlans.length); const runSpecs = resolveReportRunSpecs(params.args, params.runPlans, { concurrency }); const runVitest = params.runVitestJsonReport ?? runVitestJsonReport; - const results = []; - results.length = runSpecs.length; - const runs = []; - runs.length = runSpecs.length; - let nextIndex = 0; let failed = false; let exitCode = 0; - async function worker() { - while (nextIndex < runSpecs.length && exitCode === 0) { - const index = nextIndex; - nextIndex += 1; - const plan = runSpecs[index]; + const results = await pMap( + runSpecs, + async (plan) => { + if (exitCode !== 0) { + return null; + } const slug = sanitizePathSegment(plan.label); const run = await runVitest({ config: plan.config, @@ -1006,7 +1003,6 @@ export async function runReportPlans(params) { killGraceMs: params.args.killGraceMs, vitestArgs: plan.vitestArgs, }); - runs[index] = run; printRunLine(run); let includeEntry = true; if (run.status !== 0) { @@ -1035,24 +1031,19 @@ export async function runReportPlans(params) { } } const entry = includeEntry ? { config: plan.label, reportPath: run.reportPath, run } : null; - results[index] = entry; if (entry) { printSlowTestsForRun(entry, params.args.maxTestMs); } - } - } - - await Promise.all( - Array.from({ length: concurrency }, async () => { - await worker(); - }), + return { entry, run }; + }, + { concurrency, stopOnError: true }, ); return { failed, exitCode, - runEntries: results.filter(Boolean), - runs: runs.filter(Boolean), + runEntries: results.flatMap((result) => (result?.entry ? [result.entry] : [])), + runs: results.flatMap((result) => (result ? [result.run] : [])), }; } diff --git a/scripts/test-projects.mjs b/scripts/test-projects.mjs index 20e2405da732..39fbb6595134 100644 --- a/scripts/test-projects.mjs +++ b/scripts/test-projects.mjs @@ -2,6 +2,7 @@ // full local suite. import fs from "node:fs"; import { performance } from "node:perf_hooks"; +import pMap from "p-map"; import { formatMs } from "./lib/check-timing-summary.mjs"; import { acquireLocalHeavyCheckLockSync } from "./lib/local-heavy-check-runtime.mjs"; import { @@ -207,21 +208,21 @@ function printNoChangedTestTargets(args, cwd, baseEnv) { } async function runVitestSpecsParallel(specs, concurrency) { - let nextIndex = 0; let exitCode = 0; + let stopScheduling = false; const failures = []; const timings = []; - const runWorker = async () => { - for (;;) { - const index = nextIndex; - nextIndex += 1; - const spec = specs[index]; - if (!spec) { + await pMap( + specs, + async (spec, index) => { + if (stopScheduling) { return; } const result = await runLoggedVitestSpec(spec); if (!result) { + // A forwarded termination signal must not admit replacement shards during shutdown. + stopScheduling = true; return; } if (result.code !== 0) { @@ -238,10 +239,9 @@ async function runVitestSpecsParallel(specs, concurrency) { if (result.timing) { timings.push(result.timing); } - } - }; - - await Promise.all(Array.from({ length: concurrency }, () => runWorker())); + }, + { concurrency, stopOnError: true }, + ); return { exitCode, failures, timings }; } diff --git a/src/gateway/server-startup-post-attach.ts b/src/gateway/server-startup-post-attach.ts index f4d0df374f50..5fd26583a2ff 100644 --- a/src/gateway/server-startup-post-attach.ts +++ b/src/gateway/server-startup-post-attach.ts @@ -2,6 +2,7 @@ // Schedules warmups, sentinels, update checks, memory backend, and plugin services. import { monitorEventLoopDelay, performance } from "node:perf_hooks"; import { setTimeout as sleep } from "node:timers/promises"; +import pMap from "p-map"; import type { CliDeps } from "../cli/deps.types.js"; import { resolveStateDir } from "../config/paths.js"; import type { GatewayTailscaleMode } from "../config/types.gateway.js"; @@ -448,7 +449,6 @@ async function cleanupStaleSessionLocks(params: { Math.floor(params.concurrency ?? SESSION_LOCK_CLEANUP_CONCURRENCY), ), ); - let nextIndex = 0; let markRestartAbortedMainSessionsFromLocks = params.markRestartAbortedMainSessionsFromLocks ?? null; const getMarker = async () => { @@ -456,11 +456,10 @@ async function cleanupStaleSessionLocks(params: { .markRestartAbortedMainSessionsFromLocks; return markRestartAbortedMainSessionsFromLocks; }; - const worker = async () => { - while (!params.isStopped()) { - const sessionsDir = params.sessionDirs[nextIndex]; - nextIndex += 1; - if (!sessionsDir) { + await pMap( + params.sessionDirs, + async (sessionsDir) => { + if (params.isStopped()) { return; } const result = await params.cleanStaleLockFiles({ @@ -470,16 +469,16 @@ async function cleanupStaleSessionLocks(params: { log: { warn: (message) => params.log.warn(message) }, }); if (result.cleaned.length === 0) { - continue; + return; } const markRestartAbortedMainSessionsFromLocksLocal = await getMarker(); await markRestartAbortedMainSessionsFromLocksLocal({ sessionsDir, cleanedLocks: result.cleaned, }); - } - }; - await Promise.all(Array.from({ length: concurrency }, () => worker())); + }, + { concurrency, stopOnError: true }, + ); } function scheduleTranscriptsAutoStartSidecar(params: { diff --git a/src/infra/package-dist-inventory.ts b/src/infra/package-dist-inventory.ts index a50a9bfbd6cf..ec97685a2b3d 100644 --- a/src/infra/package-dist-inventory.ts +++ b/src/infra/package-dist-inventory.ts @@ -2,6 +2,7 @@ import fs from "node:fs/promises"; import path from "node:path"; import { sortUniqueStrings } from "@openclaw/normalization-core/string-normalization"; +import pLimit, { type LimitFunction } from "p-limit"; import { isLocalBuildMetadataDistPath } from "../../scripts/lib/local-build-metadata-paths.mjs"; import { readJsonIfExists, writeJson } from "./json-files.js"; @@ -91,38 +92,6 @@ type PackageDistInventoryRules = { externalizedExtensionIds: ExternalizedBundledExtensionIds; exclusions: PackageDistExclusionRules; }; -type PackageDistInventoryScanContext = { - activeFsOps: number; - fsConcurrency: number; - waiters: Array<() => void>; -}; - -function createPackageDistInventoryScanContext(): PackageDistInventoryScanContext { - return { - activeFsOps: 0, - fsConcurrency: PACKAGE_DIST_INVENTORY_SCAN_CONCURRENCY, - waiters: [], - }; -} - -async function withPackageDistInventoryFsSlot( - context: PackageDistInventoryScanContext, - task: () => Promise, -): Promise { - while (context.activeFsOps >= context.fsConcurrency) { - await new Promise((resolve) => { - context.waiters.push(resolve); - }); - } - context.activeFsOps += 1; - try { - return await task(); - } finally { - context.activeFsOps -= 1; - context.waiters.shift()?.(); - } -} - function normalizeRelativePath(value: string): string { return value.replace(/\\/g, "/"); } @@ -349,22 +318,20 @@ async function collectRelativeFiles( rootDir: string, baseDir: string, rules: PackageDistInventoryRules, - context: PackageDistInventoryScanContext, + fsLimit: LimitFunction, ): Promise { const rootRelativePath = normalizeRelativePath(path.relative(baseDir, rootDir)); if (rootRelativePath && isOmittedDistSubtree(rootRelativePath, rules)) { return []; } try { - const rootStats = await withPackageDistInventoryFsSlot(context, () => fs.lstat(rootDir)); + const rootStats = await fsLimit(() => fs.lstat(rootDir)); if (!rootStats.isDirectory() || rootStats.isSymbolicLink()) { throw new Error( `Unsafe package dist path: ${normalizeRelativePath(path.relative(baseDir, rootDir))}`, ); } - const entries = await withPackageDistInventoryFsSlot(context, () => - fs.readdir(rootDir, { withFileTypes: true }), - ); + const entries = await fsLimit(() => fs.readdir(rootDir, { withFileTypes: true })); const files = await Promise.all( entries.map(async (entry) => { const entryPath = path.join(rootDir, entry.name); @@ -373,7 +340,7 @@ async function collectRelativeFiles( throw new Error(`Unsafe package dist path: ${relativePath}`); } if (entry.isDirectory()) { - return await collectRelativeFiles(entryPath, baseDir, rules, context); + return await collectRelativeFiles(entryPath, baseDir, rules, fsLimit); } if (entry.isFile()) { return isPackagedDistPath(relativePath, rules) ? [relativePath] : []; @@ -393,13 +360,8 @@ async function collectRelativeFiles( /** Collects package dist files that should be present after install/update publication. */ export async function collectPackageDistInventory(packageRoot: string): Promise { const rules = await collectPackageDistInventoryRulesForRoot(packageRoot); - const scanContext = createPackageDistInventoryScanContext(); - return await collectRelativeFiles( - path.join(packageRoot, "dist"), - packageRoot, - rules, - scanContext, - ); + const fsLimit = pLimit(PACKAGE_DIST_INVENTORY_SCAN_CONCURRENCY); + return await collectRelativeFiles(path.join(packageRoot, "dist"), packageRoot, rules, fsLimit); } /** Lists legacy plugin dependency staging directories that must not ship in package dist. */ diff --git a/test/scripts/run-oxlint.test.ts b/test/scripts/run-oxlint.test.ts index 9c7859f9157f..ff301f2164ca 100644 --- a/test/scripts/run-oxlint.test.ts +++ b/test/scripts/run-oxlint.test.ts @@ -94,14 +94,6 @@ describe("run-oxlint", () => { expect(childSkipIndex).toBeGreaterThan(lockIndex); }); - it("keeps a serial oxlint shard path available", () => { - const shardedLintRunner = readFileSync("scripts/run-oxlint-shards.mjs", "utf8"); - - expect(shardedLintRunner).toContain("OPENCLAW_OXLINT_SHARDS_SERIAL"); - expect(shardedLintRunner).toContain('platform === "win32"'); - expect(shardedLintRunner).toContain("runShardsSerial"); - }); - it("serializes broad oxlint shards on constrained local hosts", () => { expect( shouldRunOxlintShardsSerial({ diff --git a/test/scripts/test-extension.test.ts b/test/scripts/test-extension.test.ts index fff5c01f8708..9bf1f3365000 100644 --- a/test/scripts/test-extension.test.ts +++ b/test/scripts/test-extension.test.ts @@ -39,6 +39,29 @@ type RunGroupParams = { env: Record; targets: string[]; }; + +function createConcurrentExtensionBatchPlan() { + const groups = [ + ["light", 10, "one", 1], + ["heavy", 30, "two", 3], + ["middle", 20, "three", 2], + ] as const; + return { + extensionCount: groups.length, + extensionIds: groups.map(([, , extensionId]) => extensionId), + estimatedCost: 60, + hasTests: true, + planGroups: groups.map(([config, estimatedCost, extensionId, testFileCount]) => ({ + config, + estimatedCost, + extensionIds: [extensionId], + roots: [`extensions/${extensionId}`], + testFileCount, + })), + testFileCount: 6, + }; +} + function runScriptResult(args: string[], cwd = process.cwd()) { return spawnSync(process.execPath, [scriptPath, ...args], { cwd, @@ -561,48 +584,17 @@ describe("scripts/test-extension.mjs", () => { resolvers.push(() => resolve(0)); }); }); - const runPromise = runExtensionBatchPlan( - { - extensionCount: 3, - extensionIds: ["one", "two", "three"], - estimatedCost: 60, - hasTests: true, - planGroups: [ - { - config: "light", - estimatedCost: 10, - extensionIds: ["one"], - roots: ["extensions/one"], - testFileCount: 1, - }, - { - config: "heavy", - estimatedCost: 30, - extensionIds: ["two"], - roots: ["extensions/two"], - testFileCount: 3, - }, - { - config: "middle", - estimatedCost: 20, - extensionIds: ["three"], - roots: ["extensions/three"], - testFileCount: 2, - }, - ], - testFileCount: 6, - }, - { - env: { OPENCLAW_EXTENSION_BATCH_PARALLEL: "2" }, - runGroup: runGroup as NonNullable< - NonNullable[1]>["runGroup"] - >, - vitestArgs: ["--reporter=dot"], - }, - ); + const runPromise = runExtensionBatchPlan(createConcurrentExtensionBatchPlan(), { + env: { OPENCLAW_EXTENSION_BATCH_PARALLEL: "2" }, + runGroup: runGroup as NonNullable< + NonNullable[1]>["runGroup"] + >, + vitestArgs: ["--reporter=dot"], + }); - await Promise.resolve(); - expect(started).toEqual(["heavy", "middle"]); + await vi.waitFor(() => { + expect(started).toEqual(["heavy", "middle"]); + }); resolvers.shift()?.(); await new Promise((resolve) => { setImmediate(resolve); @@ -631,6 +623,40 @@ describe("scripts/test-extension.mjs", () => { }); }); + it("stops admitting extension batch groups after a parallel failure", async () => { + const started: string[] = []; + let resolveHeavy: ((code: number) => void) | undefined; + let resolveMiddle: ((code: number) => void) | undefined; + const runGroup = vi.fn((params: RunGroupParams) => { + started.push(params.config); + return new Promise((resolve) => { + if (params.config === "heavy") { + resolveHeavy = resolve; + } else if (params.config === "middle") { + resolveMiddle = resolve; + } + }); + }); + const runPromise = runExtensionBatchPlan(createConcurrentExtensionBatchPlan(), { + env: { OPENCLAW_EXTENSION_BATCH_PARALLEL: "2" }, + runGroup: runGroup as NonNullable< + NonNullable[1]>["runGroup"] + >, + }); + + await vi.waitFor(() => { + expect(started).toEqual(["heavy", "middle"]); + }); + resolveHeavy?.(7); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(started).toEqual(["heavy", "middle"]); + resolveMiddle?.(0); + await expect(runPromise).resolves.toBe(7); + expect(runGroup).toHaveBeenCalledTimes(2); + }); + it("keeps extension batch parallelism bounded by group count", () => { expect(resolveExtensionBatchParallelism(3, { OPENCLAW_EXTENSION_BATCH_PARALLEL: "2" })).toBe(2); expect(resolveExtensionBatchParallelism(1, { OPENCLAW_EXTENSION_BATCH_PARALLEL: "4" })).toBe(1); diff --git a/test/scripts/test-group-report.test.ts b/test/scripts/test-group-report.test.ts index 0ec65acd1711..0b2aaec86757 100644 --- a/test/scripts/test-group-report.test.ts +++ b/test/scripts/test-group-report.test.ts @@ -400,6 +400,67 @@ describe("scripts/test-group-report aggregation", () => { } }); + it("stops admitting report plans after a parallel failure", async () => { + const tempDir = makeTempDir(); + const labels = ["first", "second", "third"]; + const started: string[] = []; + const resolvers = new Map void>(); + try { + const runPromise = runReportPlans({ + args: parseTestGroupReportArgs([ + ...labels.flatMap((label) => ["--config", `${label}.config.ts`]), + "--concurrency", + "2", + "--no-rss", + ]), + logDir: path.join(tempDir, "logs"), + reportDir: path.join(tempDir, "reports"), + runPlans: labels.map((label) => ({ + config: `${label}.config.ts`, + forwardedArgs: [], + label, + })), + runVitestJsonReport: async (params: { + config: string; + label: string; + logPath: string; + reportPath: string; + }) => { + started.push(params.label); + const status = await new Promise((resolve) => { + resolvers.set(params.label, resolve); + }); + return { + config: params.config, + elapsedMs: 10, + label: params.label, + logPath: params.logPath, + maxRssBytes: null, + reportPath: params.reportPath, + status, + }; + }, + }); + + await vi.waitFor(() => { + expect(started).toStrictEqual(["first", "second"]); + }); + resolvers.get("first")?.(1); + await new Promise((resolve) => { + setImmediate(resolve); + }); + expect(started).toStrictEqual(["first", "second"]); + resolvers.get("second")?.(0); + + const result = await runPromise; + expect(result.exitCode).toBe(1); + expect(result.failed).toBe(true); + expect(result.runs.map((run) => run.label)).toStrictEqual(["first", "second"]); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); + } + }); + it("prints slow tests as soon as each config report completes", async () => { const tempDir = makeTempDir(); const logSpy = vi.spyOn(console, "log").mockImplementation(() => {});