diff --git a/packages/tooling/task-runner/__tests__/integration/e2e-runner.test.ts b/packages/tooling/task-runner/__tests__/integration/e2e-runner.test.ts index 7b682d56..dbab4373 100644 --- a/packages/tooling/task-runner/__tests__/integration/e2e-runner.test.ts +++ b/packages/tooling/task-runner/__tests__/integration/e2e-runner.test.ts @@ -4,7 +4,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { promisify } from "node:util"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { defaultTaskRunner } from "../../src/default-task-runner"; import { EmptyLifeCycle } from "../../src/life-cycle"; @@ -125,7 +125,7 @@ describe("defaultTaskRunner (E2E via real subprocess)", () => { }); afterEach(async () => { - await rm(workspaceRoot, { force: true, recursive: true }); + await rm(workspaceRoot, { force: true, maxRetries: 20, recursive: true, retryDelay: 100 }); }); it("round-trips a real subprocess through the cache (cold miss → warm hit restores output)", async () => { @@ -350,3 +350,62 @@ describe("defaultTaskRunner (E2E via real subprocess)", () => { expect(checkCount).toBe(2); }); }); + +describe("remote cache failures degrade to a miss", () => { + let workspaceRoot: string; + + beforeEach(async () => { + workspaceRoot = await createTemporaryDirectory(); + + await mkdir(join(workspaceRoot, "packages/app/src"), { recursive: true }); + await writeFile(join(workspaceRoot, "packages/app/src/index.ts"), "export const x = 1;\n"); + await writeFile(join(workspaceRoot, "packages/app/package.json"), JSON.stringify({ name: "app" })); + }); + + afterEach(async () => { + await rm(workspaceRoot, { force: true, maxRetries: 20, recursive: true, retryDelay: 100 }); + }); + + it("runs the tasks normally and warns once when the remote cache is unreachable", async () => { + expect.assertions(5); + + const build = makeTask("app:build", "packages/app"); + const check = makeTask("app:check", "packages/app"); + + const executor = createNodeExecutor(workspaceRoot, { + "app:build": "process.stdout.write('built');", + "app:check": "process.stdout.write('checked');", + }); + const context = createContext(workspaceRoot, [build, check], { "app:build": [], "app:check": [] }, executor); + const diagnostics: { message: string; taskId: string }[] = []; + const onUploadError = vi.fn<(hash: string, error: unknown) => void>(); + + const results = await defaultTaskRunner( + [build, check], + { + onDiagnostic: (taskId, message) => { + diagnostics.push({ message, taskId }); + }, + remoteCache: { + mode: "read", + onUploadError, + timeout: 100, + url: "http://127.0.0.1:1", + }, + }, + context, + ); + + expect(results.get("app:build")?.status).toBe("success"); + expect(results.get("app:check")?.status).toBe("success"); + + const remoteWarnings = diagnostics.filter((entry) => entry.taskId === "remote-cache"); + + expect(remoteWarnings).toHaveLength(1); + expect(remoteWarnings[0]?.message).toMatch(/Remote cache unavailable/); + + // The caller's detail hook still sees every failed operation, so a + // debug logger can retain the full sequence. + expect(onUploadError).toHaveBeenCalledTimes(2); + }); +}); diff --git a/packages/tooling/task-runner/__tests__/unit/affected-external-inputs.test.ts b/packages/tooling/task-runner/__tests__/unit/affected-external-inputs.test.ts new file mode 100644 index 00000000..0aa700bd --- /dev/null +++ b/packages/tooling/task-runner/__tests__/unit/affected-external-inputs.test.ts @@ -0,0 +1,150 @@ +import { execFileSync } from "node:child_process"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import { getAffectedProjects } from "../../src/affected"; +import { collectExternalInputs, computeExternalInputHashes } from "../../src/external-inputs"; +import type { ProjectConfiguration, ProjectGraph } from "../../src/types"; + +const git = (cwd: string, ...arguments_: string[]): void => { + execFileSync("git", arguments_, { cwd, stdio: "pipe" }); +}; + +describe("affected detection with external inputs", () => { + let workspaceRoot: string; + let externalRoot: string; + let externalFile: string; + + beforeEach(async () => { + const base = await mkdtemp(join(tmpdir(), "tr-affected-external-")); + + workspaceRoot = join(base, "workspace"); + externalRoot = join(base, "shared"); + + await mkdir(join(workspaceRoot, "packages/lib-a"), { recursive: true }); + await mkdir(join(workspaceRoot, "packages/app"), { recursive: true }); + await mkdir(externalRoot, { recursive: true }); + + await writeFile(join(workspaceRoot, "packages/lib-a/index.ts"), "export const a = 1;\n"); + await writeFile(join(workspaceRoot, "packages/app/index.ts"), "export const app = 1;\n"); + + externalFile = join(externalRoot, "constants.ts"); + await writeFile(externalFile, "export const shared = 1;\n"); + + git(workspaceRoot, "init", "--initial-branch=main"); + git(workspaceRoot, "config", "user.email", "test@example.com"); + git(workspaceRoot, "config", "user.name", "Test"); + git(workspaceRoot, "add", "."); + git(workspaceRoot, "commit", "-m", "initial"); + }); + + afterEach(async () => { + await rm(join(workspaceRoot, ".."), { force: true, recursive: true }); + }); + + it("hashes external glob inputs and detects a change between snapshots", async () => { + expect.assertions(4); + + const globPattern = `${externalRoot}/**/*.ts`; + const externalInputs = [{ pattern: globPattern, projects: ["lib-a"] }]; + const projects: Record = { "lib-a": { root: "packages/lib-a" } }; + const projectGraph: ProjectGraph = { + dependencies: { "lib-a": [] }, + nodes: { "lib-a": { data: { root: "packages/lib-a" }, name: "lib-a", type: "library" } }, + }; + + const before = await computeExternalInputHashes(externalInputs); + + expect(before[globPattern]).toBeTypeOf("string"); + + await writeFile(join(externalRoot, "extra.ts"), "export const extra = 1;\n"); + + const after = await computeExternalInputHashes(externalInputs); + + expect(after[globPattern]).not.toBe(before[globPattern]); + + const firstRun = await getAffectedProjects({ + base: "HEAD", + externalInputs, + head: "HEAD", + projectGraph, + projects, + workspaceRoot, + }); + + // No previous snapshot: the owning project must not be reported clean. + expect(firstRun.changedExternalInputs).toStrictEqual([globPattern]); + + const secondRun = await getAffectedProjects({ + base: "HEAD", + externalInputs, + head: "HEAD", + previousExternalInputHashes: firstRun.externalInputHashes, + projectGraph, + projects, + workspaceRoot, + }); + + expect(secondRun.changedExternalInputs).toStrictEqual([]); + }); + + it("marks the owning project affected when an external input changes and its dependents downstream", async () => { + expect.assertions(4); + + const externalInputs = collectExternalInputs(workspaceRoot, { + "lib-a": { root: "packages/lib-a", targets: { build: { inputs: [`file://${externalFile}`] } } }, + }); + const graph: ProjectGraph = { + dependencies: { app: [{ source: "app", target: "lib-a", type: "static" }], "lib-a": [] }, + nodes: { + app: { data: { root: "packages/app" }, name: "app", type: "application" }, + "lib-a": { data: { root: "packages/lib-a" }, name: "lib-a", type: "library" }, + }, + }; + const projects: Record = { + app: { root: "packages/app" }, + "lib-a": { root: "packages/lib-a" }, + }; + + const baseline = await getAffectedProjects({ + base: "HEAD", + externalInputs, + head: "HEAD", + projectGraph: graph, + projects, + workspaceRoot, + }); + + expect(baseline.changedProjects).toStrictEqual(["lib-a"]); + + const unchanged = await getAffectedProjects({ + base: "HEAD", + externalInputs, + head: "HEAD", + previousExternalInputHashes: baseline.externalInputHashes, + projectGraph: graph, + projects, + workspaceRoot, + }); + + expect(unchanged.changedProjects).toStrictEqual([]); + + await writeFile(externalFile, "export const shared = 3;\n"); + + const changed = await getAffectedProjects({ + base: "HEAD", + externalInputs, + head: "HEAD", + previousExternalInputHashes: baseline.externalInputHashes, + projectGraph: graph, + projects, + workspaceRoot, + }); + + expect(changed.changedProjects).toStrictEqual(["lib-a"]); + expect(changed.affectedProjects).toStrictEqual(["lib-a", "app"]); + }); +}); diff --git a/packages/tooling/task-runner/__tests__/unit/backends/http.test.ts b/packages/tooling/task-runner/__tests__/unit/backends/http.test.ts index a6f3b38b..a320a1fc 100644 --- a/packages/tooling/task-runner/__tests__/unit/backends/http.test.ts +++ b/packages/tooling/task-runner/__tests__/unit/backends/http.test.ts @@ -5,9 +5,11 @@ import type { IncomingMessage, Server, ServerResponse } from "node:http"; import { createServer } from "node:http"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { Readable } from "node:stream"; import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { createRemoteCacheBackend } from "../../../src/backends/factory"; import { actionDigestForTaskHash, containsByTaskHash, retrieveByTaskHash, storeByTaskHash } from "../../../src/backends/hash-bridge"; import { HttpRemoteCache } from "../../../src/backends/http"; @@ -1197,6 +1199,83 @@ describe(HttpRemoteCache, () => { }); }); + describe("remote cache error resilience", () => { + it("treats a 502 read as a miss and reports each failure without throwing", async () => { + expect.assertions(3); + + const { server, url } = await startMockServer((_request, response) => { + response.writeHead(502); + response.end(); + }); + + try { + const errors: unknown[] = []; + const cache = createRemoteCacheBackend({ localCasRoot: cacheDirectory, timeout: 250, url }, (error) => errors.push(error)); + + const first = await retrieveByTaskHash(cache, "first", cacheDirectory); + const second = await retrieveByTaskHash(cache, "second", cacheDirectory); + + expect(first).toBe(false); + expect(second).toBe(false); + expect(errors).toHaveLength(2); + + await cache.close(); + } finally { + await closeServer(server); + } + }); + + it("falls back to the caller's onUploadError hook when no reporter is supplied", async () => { + expect.assertions(2); + + const { server, url } = await startMockServer((_request, response) => { + response.writeHead(500); + response.end(); + }); + + try { + const onUploadError = vi.fn<(hash: string, error: unknown) => void>(); + const cache = createRemoteCacheBackend({ localCasRoot: cacheDirectory, onUploadError, timeout: 250, url }); + + await retrieveByTaskHash(cache, "first", cacheDirectory); + await retrieveByTaskHash(cache, "second", cacheDirectory); + + expect(onUploadError).toHaveBeenCalledTimes(2); + expect(onUploadError.mock.calls[0]?.[0]).toBe("remote-cache"); + } finally { + await closeServer(server); + } + }); + + it("treats a 502 write as a failed upload without throwing", async () => { + expect.assertions(2); + + const { server, url } = await startMockServer((_request, response) => { + response.writeHead(502); + response.end(); + }); + + try { + const errors: unknown[] = []; + const cache = createRemoteCacheBackend({ timeout: 250, url }, (error) => errors.push(error)); + const bytes = Buffer.from("payload", "utf8"); + const digest = { hash: createHash("sha256").update(bytes).digest("hex"), sizeBytes: bytes.byteLength }; + const stored = await cache.storeAction( + actionDigestForTaskHash("write-502"), + { exitCode: 0, outputDirectories: [], outputFiles: [{ digest, isExecutable: false, path: "vis-entry.tar.gz" }] }, + [{ digest, open: () => Promise.resolve(Readable.from(bytes)) }], + ); + + expect(stored).toBe(false); + expect(errors).toHaveLength(1); + + await cache.close(); + } finally { + await closeServer(server); + } + }); + }); + describe(actionDigestForTaskHash, () => { it("derives a stable sha256 digest from the task hash", () => { expect.assertions(3); diff --git a/packages/tooling/task-runner/__tests__/unit/backends/reapi.test.ts b/packages/tooling/task-runner/__tests__/unit/backends/reapi.test.ts index ab292801..75f023b1 100644 --- a/packages/tooling/task-runner/__tests__/unit/backends/reapi.test.ts +++ b/packages/tooling/task-runner/__tests__/unit/backends/reapi.test.ts @@ -4,7 +4,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { Readable } from "node:stream"; -import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { actionDigestForTaskHash, containsByTaskHash, retrieveByTaskHash, storeByTaskHash } from "../../../src/backends/hash-bridge"; import { ReapiRemoteCache } from "../../../src/backends/reapi"; @@ -422,6 +422,27 @@ describe(ReapiRemoteCache, () => { expect(result).toBeNull(); }); + + it("treats a non-NOT_FOUND gRPC error as a cache miss through the hash bridge", async () => { + expect.assertions(3); + + fixture.server.getActionResultErrorOverride = { code: 13, message: "internal" }; + + const cache = new ReapiRemoteCache({ url: fixture.server.address }); + const directory = await createTemporaryDirectory(); + const onError = vi.fn<(hash: string, error: unknown) => void>(); + + try { + // The backend method still surfaces the transport error; the + // orchestrator's bridge is where a failed read becomes a miss + // so the task can run normally. + await expect(cache.retrieveAction({ hash: "c".repeat(64), sizeBytes: 1 })).rejects.toMatchObject({ code: 13 }); + await expect(retrieveByTaskHash(cache, "unreachable", directory, onError)).resolves.toBe(false); + expect(onError).toHaveBeenCalledTimes(1); + } finally { + await rm(directory, { force: true, recursive: true }); + } + }); }); describe("containsAction", () => { diff --git a/packages/tooling/task-runner/__tests__/unit/cache.test.ts b/packages/tooling/task-runner/__tests__/unit/cache.test.ts index 32c2c46b..c979f13c 100644 --- a/packages/tooling/task-runner/__tests__/unit/cache.test.ts +++ b/packages/tooling/task-runner/__tests__/unit/cache.test.ts @@ -2,7 +2,7 @@ import { mkdir, readFile, rm, stat, utimes, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { ActionResult } from "../../src/backends/types"; import { Cache, formatCacheSize, parseCacheSize } from "../../src/cache"; @@ -102,21 +102,23 @@ describe(Cache, () => { it("surfaces write failures through onWriteError without throwing", async () => { expect.assertions(3); - // Point the cache directory at a regular file so every staging - // mkdir inside put() fails (ENOTDIR). put() must stay non-throwing - // (best-effort caching) but report the failure so a broken cache - // isn't silently invisible. - const notADirectory = join(workspaceRoot, "cache-is-a-file"); - - await writeFile(notADirectory, "x"); + // Let the constructor create the cache directory, then replace it + // with a regular file so the per-entry staging mkdir inside put() + // fails (ENOTDIR). put() must stay non-throwing (best-effort + // caching) but report the failure so a broken cache isn't + // silently invisible. + const cacheDirectory = join(workspaceRoot, "cache-is-a-file"); const failures: { error: unknown; hash: string }[] = []; const failingCache = new Cache({ - cacheDirectory: notADirectory, + cacheDirectory, onWriteError: (hash, error) => failures.push({ error, hash }), workspaceRoot, }); + await rm(cacheDirectory, { force: true, recursive: true }); + await writeFile(cacheDirectory, "x"); + await expect(failingCache.put("boom", "output", [], 0)).resolves.toBeUndefined(); expect(failures).toHaveLength(1); @@ -662,3 +664,44 @@ describe(formatCacheSize, () => { expect(formatCacheSize(2 * 1024 * 1024 * 1024)).toBe("2.0GB"); }); }); + +describe("cache directory setup", () => { + it("disables the cache instead of throwing when the directory cannot be created", async () => { + expect.assertions(8); + + const workspaceRoot = await createTemporaryDirectory(); + const blockingPath = join(workspaceRoot, "not-a-directory"); + + await writeFile(blockingPath, "a regular file blocks mkdir"); + + const onSetupError = vi.fn<(directory: string, error: unknown) => void>(); + let cache: Cache | undefined; + + // Constructing is side-effect free: read-only callers (cache + // inspection, `cache clean`) must not materialise directories. + expect(() => { + cache = new Cache({ cacheDirectory: blockingPath, onSetupError, workspaceRoot }); + }).not.toThrow(); + + const disabledCache = cache as Cache; + + expect(disabledCache.disabled).toBe(false); + + await disabledCache.ensureReady(); + + expect(disabledCache.disabled).toBe(true); + expect(onSetupError).toHaveBeenCalledTimes(1); + + const [failedDirectory, error] = onSetupError.mock.calls[0] as [string, NodeJS.ErrnoException]; + + expect(failedDirectory).toBe(blockingPath); + expect(error.code).toMatch(/EEXIST|ENOTDIR/); + + // Every operation degrades to a miss/no-op rather than throwing, so + // the run continues without caching. + await expect(disabledCache.put("hash", "output", [], 0)).resolves.toBeUndefined(); + await expect(disabledCache.get("hash")).resolves.toBeUndefined(); + + await rm(workspaceRoot, { force: true, recursive: true }); + }); +}); diff --git a/packages/tooling/task-runner/__tests__/unit/default-task-runner.test.ts b/packages/tooling/task-runner/__tests__/unit/default-task-runner.test.ts index cbb21e33..38178696 100644 --- a/packages/tooling/task-runner/__tests__/unit/default-task-runner.test.ts +++ b/packages/tooling/task-runner/__tests__/unit/default-task-runner.test.ts @@ -308,3 +308,46 @@ describe(defaultTaskRunner, () => { expect(getCount()).toBe(2); }); }); + +describe("defaultTaskRunner cache directory setup", () => { + it("runs the task and warns once when the cache directory cannot be created", async () => { + expect.assertions(4); + + const workspaceRoot = await createTemporaryDirectory(); + + await mkdir(join(workspaceRoot, "packages/app/src"), { recursive: true }); + await writeFile(join(workspaceRoot, "packages/app/src/index.ts"), "const x = 1;"); + + const blockingPath = join(workspaceRoot, "cache-is-a-file"); + + await writeFile(blockingPath, "not a directory"); + + const task = makeTask(); + const { executor, getCount } = createCountingExecutor(); + const context = createContext(workspaceRoot, [task], executor); + const diagnostics: { message: string; taskId: string }[] = []; + + try { + const results = await defaultTaskRunner( + [task], + { + cacheDirectory: blockingPath, + onDiagnostic: (taskId, message) => { + diagnostics.push({ message, taskId }); + }, + }, + context, + ); + + expect(results.get("app:build")?.status).toBe("success"); + expect(getCount()).toBe(1); + + const warnings = diagnostics.filter((entry) => entry.taskId === "cache"); + + expect(warnings).toHaveLength(1); + expect(warnings[0]?.message).toContain(blockingPath); + } finally { + await rm(workspaceRoot, { force: true, recursive: true }); + } + }); +}); diff --git a/packages/tooling/task-runner/__tests__/unit/external-inputs.test.ts b/packages/tooling/task-runner/__tests__/unit/external-inputs.test.ts new file mode 100644 index 00000000..432f4cf8 --- /dev/null +++ b/packages/tooling/task-runner/__tests__/unit/external-inputs.test.ts @@ -0,0 +1,103 @@ +import { execFileSync } from "node:child_process"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, relative } from "node:path"; + +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import { collectExternalInputs, isExternalPattern } from "../../src/external-inputs"; +import { InProcessTaskHasher } from "../../src/task-hasher"; +import type { ProjectConfiguration, Task } from "../../src/types"; + +const task: Task = { + cache: true, + id: "lib-a:build", + outputs: [], + overrides: {}, + target: { project: "lib-a", target: "build" }, +}; + +const git = (cwd: string, ...arguments_: string[]): void => { + execFileSync("git", arguments_, { cwd, stdio: "pipe" }); +}; + +describe("external file-set inputs", () => { + let workspaceRoot: string; + let externalRoot: string; + let externalFile: string; + + beforeEach(async () => { + const base = await mkdtemp(join(tmpdir(), "tr-external-")); + + workspaceRoot = join(base, "workspace"); + externalRoot = join(base, "shared"); + + await mkdir(join(workspaceRoot, "packages/lib-a"), { recursive: true }); + await mkdir(externalRoot, { recursive: true }); + await writeFile(join(workspaceRoot, "packages/lib-a/index.ts"), "export const a = 1;\n"); + + externalFile = join(externalRoot, "constants.ts"); + await writeFile(externalFile, "export const shared = 1;\n"); + + git(workspaceRoot, "init", "--initial-branch=main"); + git(workspaceRoot, "config", "user.email", "test@example.com"); + git(workspaceRoot, "config", "user.name", "Test"); + git(workspaceRoot, "add", "."); + git(workspaceRoot, "commit", "-m", "initial"); + }); + + afterEach(async () => { + await rm(join(workspaceRoot, ".."), { force: true, recursive: true }); + }); + + it("collects patterns that resolve outside the workspace and attributes them to their project", async () => { + expect.assertions(3); + + const projects: Record = { + "lib-a": { + root: "packages/lib-a", + targets: { build: { inputs: [`file://${externalFile}`, "glob://{workspaceRoot}/../shared/**/*.ts", "{projectRoot}/**/*"] } }, + }, + }; + + const externalInputs = collectExternalInputs(workspaceRoot, projects); + + expect(externalInputs).toHaveLength(2); + expect(externalInputs.map((input) => input.pattern).toSorted()).toStrictEqual([externalFile, join(externalRoot, "**/*.ts")].toSorted()); + expect(externalInputs.every((input) => input.projects.join(",") === "lib-a")).toBe(true); + }); + + it("treats an absolute path outside the workspace as external and one inside as local", () => { + expect.assertions(2); + + expect(isExternalPattern(workspaceRoot, externalFile)).toBe(true); + expect(isExternalPattern(workspaceRoot, join(workspaceRoot, "packages/lib-a/index.ts"))).toBe(false); + }); + + it("folds an external file into the task hash with a workspace-relative key", async () => { + expect.assertions(3); + + const hasher = new InProcessTaskHasher({ + projects: { + "lib-a": { + root: "packages/lib-a", + targets: { build: { inputs: [`file://${externalFile}`] } }, + }, + }, + workspaceRoot, + }); + + const first = await hasher.hashTask(task); + const key = relative(workspaceRoot, externalFile).replaceAll("\\", "/"); + + expect(Object.keys(first.nodes)).toStrictEqual([key]); + expect(key.startsWith("..")).toBe(true); + + await writeFile(externalFile, "export const shared = 2;\n"); + hasher.clearCache(); + + const second = await hasher.hashTask(task); + + expect(second.nodes[key]).not.toBe(first.nodes[key]); + }); +}); diff --git a/packages/tooling/task-runner/src/affected.ts b/packages/tooling/task-runner/src/affected.ts index 68e63e1f..43dc2b58 100644 --- a/packages/tooling/task-runner/src/affected.ts +++ b/packages/tooling/task-runner/src/affected.ts @@ -7,6 +7,8 @@ import { existsSync } from "node:fs"; import { matchesGlob } from "node:path"; // path utilities not needed - git returns workspace-relative paths +import type { ExternalInputDefinition } from "./external-inputs"; +import { computeExternalInputHashes, findChangedExternalInputs, projectsForExternalPatterns } from "./external-inputs"; import type { AffectedScope, ProjectConfiguration, ProjectGraph } from "./types"; /** @@ -55,6 +57,14 @@ interface AffectedOptions { */ downstream?: AffectedScope; + /** + * File-set inputs that resolve outside `workspaceRoot`, together with the + * projects that declared them. Built by `collectExternalInputs` so + * affected detection maps external paths to the same owners the hasher + * used for the cache key. + */ + externalInputs?: ExternalInputDefinition[]; + /** * Owners for files that lie outside every project root but belong to * specific projects rather than the whole workspace, keyed by @@ -85,6 +95,13 @@ interface AffectedOptions { * `docs/guide.md`; use `docs/**` for a whole folder. */ ignoredFiles?: string[]; + + /** + * External-input content hashes from the previous run, keyed by pattern. + * When omitted every external input counts as changed, because affected + * detection has no baseline to compare against. + */ + previousExternalInputHashes?: Record; /** Project graph for dependency resolution */ projectGraph: ProjectGraph; /** All project configurations keyed by name */ @@ -105,12 +122,20 @@ interface AffectedOptions { interface AffectedResult { /** All affected projects (union of changed, downstream, and upstream) */ affectedProjects: string[]; + /** External input patterns whose contents changed since the previous snapshot */ + changedExternalInputs: string[]; /** Files that changed between base and head */ changedFiles: string[]; /** Projects that were directly changed */ changedProjects: string[]; /** Projects affected because they depend on changed projects */ downstreamProjects: string[]; + + /** + * Current content hashes for every external input pattern. Callers + * persist this as the next run's baseline. + */ + externalInputHashes: Record; /** Changed files outside every project that were skipped because they matched `ignoredFiles` */ ignoredFiles: string[]; /** Projects that changed projects depend on */ @@ -367,9 +392,11 @@ const getAffectedProjects = async (options: AffectedOptions): Promise 0 ? await computeExternalInputHashes(externalInputs) : {}; + const changedExternalInputs = externalInputs ? findChangedExternalInputs(externalInputs, externalInputHashes, previousExternalInputHashes) : []; + if (globalChange) { return { affectedProjects: Object.keys(projects), + changedExternalInputs, changedFiles, changedProjects: [...changedProjects], downstreamProjects: [], + externalInputHashes, ignoredFiles, upstreamProjects: [], }; } + for (const project of projectsForExternalPatterns(externalInputs ?? [], changedExternalInputs)) { + changedProjects.add(project); + } + // Walk the dependency graph with scope control const result = expandAffected(changedProjects, projectGraph, { downstream, upstream }); return { affectedProjects: [...result.affected], + changedExternalInputs, changedFiles, changedProjects: [...changedProjects], downstreamProjects: [...result.downstream], + externalInputHashes, ignoredFiles, upstreamProjects: [...result.upstream], }; diff --git a/packages/tooling/task-runner/src/backends/factory.ts b/packages/tooling/task-runner/src/backends/factory.ts index 9e3be2b3..e99f30ce 100644 --- a/packages/tooling/task-runner/src/backends/factory.ts +++ b/packages/tooling/task-runner/src/backends/factory.ts @@ -49,22 +49,78 @@ export const resolveTurboEnvCompat = (input?: Partial, env: * Construct the configured remote cache backend. Selects between the * Turborepo-compatible HTTP client and the Bazel REAPI gRPC client * via `options.backend`. + * + * The returned backend is wrapped so that a transport-level failure — a 5xx, + * a dropped connection, a timeout, any thrown error from an RPC — behaves + * like a cache miss instead of failing the run. `onError` observes every + * failure; callers that surface warnings should dedupe. Errors raised by the + * task command itself never pass through this boundary and are unaffected. */ -export const createRemoteCacheBackend = (options: RemoteCacheOptions): RemoteCacheBackend => { +export const createRemoteCacheBackend = (options: RemoteCacheOptions, onError?: (error: unknown, operation: "read" | "write") => void): RemoteCacheBackend => { const mode = resolveCacheMode(options); - const resolved: RemoteCacheOptions = { ...options, mode }; - switch (options.backend ?? "http") { - case "http": { - return new HttpRemoteCache(resolved); + const report = (error: unknown, operation: "read" | "write"): void => { + if (onError) { + onError(error, operation); + } else { + // Preserve the caller's existing upload hook when no wrapper-level + // reporter was provided. + options.onUploadError?.("remote-cache", error); } + }; - case "reapi": { - return new ReapiRemoteCache(resolved); - } + const resolved: RemoteCacheOptions = { + ...options, + mode, + // Both backends report wire-level upload failures through this hook. + // Routing it through `report` keeps those errors on the same + // once-per-run path as thrown read errors. The hook is upload-shaped, + // so a read failure surfaced here is labelled "write"; callers should + // treat that label as best-effort. + onUploadError: (_hash, error) => { + report(error, "write"); + }, + }; - default: { - return new HttpRemoteCache(resolved); - } - } + const backend: RemoteCacheBackend = options.backend === "reapi" ? new ReapiRemoteCache(resolved) : new HttpRemoteCache(resolved); + + return { + close: () => backend.close(), + containsAction: async (digest) => { + try { + return await backend.containsAction(digest); + } catch (error) { + report(error, "read"); + + return false; + } + }, + fetchBlob: async (digest, destinationPath) => { + try { + return await backend.fetchBlob(digest, destinationPath); + } catch (error) { + report(error, "read"); + + return false; + } + }, + retrieveAction: async (digest) => { + try { + return await backend.retrieveAction(digest); + } catch (error) { + report(error, "read"); + + return null; + } + }, + storeAction: async (digest, result, blobs) => { + try { + return await backend.storeAction(digest, result, blobs); + } catch (error) { + report(error, "write"); + + return false; + } + }, + }; }; diff --git a/packages/tooling/task-runner/src/backends/hash-bridge.ts b/packages/tooling/task-runner/src/backends/hash-bridge.ts index 26868f04..9a9588f0 100644 --- a/packages/tooling/task-runner/src/backends/hash-bridge.ts +++ b/packages/tooling/task-runner/src/backends/hash-bridge.ts @@ -59,10 +59,16 @@ export const actionDigestForTaskHash = (taskHash: string): CasDigest => { * the orchestrator's task hash. Resolves `false` on any wire failure — * the orchestrator treats existence checks as best-effort. */ -export const containsByTaskHash = async (backend: RemoteCacheBackend, taskHash: string): Promise => { +export const containsByTaskHash = async ( + backend: RemoteCacheBackend, + taskHash: string, + onError?: (taskHash: string, error: unknown) => void, +): Promise => { try { return await backend.containsAction(actionDigestForTaskHash(taskHash)); - } catch { + } catch (error) { + onError?.(taskHash, error); + return false; } }; @@ -73,9 +79,27 @@ export const containsByTaskHash = async (backend: RemoteCacheBackend, taskHash: * directory has been fully populated; partial-extract failures clean * up after themselves so the local cache never observes a half-populated * entry. + * + * Any backend failure — a 502, a dropped connection, a timeout, a + * transport-level gRPC status — is reported through `onError` and resolved + * as a miss. Remote caching is an optimisation: an unreachable cache must + * never fail the task that would otherwise run normally. */ -export const retrieveByTaskHash = async (backend: RemoteCacheBackend, taskHash: string, localCacheDirectory: string): Promise => { - const action = await backend.retrieveAction(actionDigestForTaskHash(taskHash)); +export const retrieveByTaskHash = async ( + backend: RemoteCacheBackend, + taskHash: string, + localCacheDirectory: string, + onError?: (taskHash: string, error: unknown) => void, +): Promise => { + let action; + + try { + action = await backend.retrieveAction(actionDigestForTaskHash(taskHash)); + } catch (error) { + onError?.(taskHash, error); + + return false; + } if (action === null || action.outputFiles.length === 0) { return false; @@ -138,9 +162,11 @@ export const retrieveByTaskHash = async (backend: RemoteCacheBackend, taskHash: } return true; - } catch { + } catch (error) { await rm(stagingDirectory, { force: true, recursive: true }).catch(() => {}); + onError?.(taskHash, error); + return false; } finally { await rm(archivePath, { force: true }).catch(() => {}); diff --git a/packages/tooling/task-runner/src/backends/http.ts b/packages/tooling/task-runner/src/backends/http.ts index 6a29a390..8be1d6b2 100644 --- a/packages/tooling/task-runner/src/backends/http.ts +++ b/packages/tooling/task-runner/src/backends/http.ts @@ -338,7 +338,13 @@ class HttpRemoteCache implements RemoteCacheBackend { outputDirectories: [], outputFiles: [{ digest: blobDigest, isExecutable: false, path: BLOB_OUTPUT_PATH }], }; - } catch { + } catch (error) { + // A connection error, timeout, or abort reached the end of the + // retry budget. Surface it through the same hook as a non-OK + // status so a persistent cache outage is visible once per run + // while the read still degrades to a miss. + this.#onUploadError?.(actionDigest.hash, error); + return null; } finally { await rm(stagingPath, { force: true }).catch(() => {}); diff --git a/packages/tooling/task-runner/src/backends/reapi.ts b/packages/tooling/task-runner/src/backends/reapi.ts index a76ca4d9..0ab8513d 100644 --- a/packages/tooling/task-runner/src/backends/reapi.ts +++ b/packages/tooling/task-runner/src/backends/reapi.ts @@ -349,8 +349,9 @@ export class ReapiRemoteCache implements RemoteCacheBackend { } return await this.#streamRead(digest, destinationPath); - } catch { + } catch (error) { await rm(destinationPath, { force: true }).catch(() => {}); + this.#onUploadError?.(digest.hash, error); return false; } diff --git a/packages/tooling/task-runner/src/backends/types.ts b/packages/tooling/task-runner/src/backends/types.ts index 8e55ca41..f943f061 100644 --- a/packages/tooling/task-runner/src/backends/types.ts +++ b/packages/tooling/task-runner/src/backends/types.ts @@ -304,9 +304,10 @@ export interface RemoteCacheOptions { mode?: CacheMode; /** - * Called when a fire-and-forget upload fails. Uploads are - * non-blocking, so without this hook errors are silently dropped. - * Provide it to log or report upload failures. + * Called when a remote cache upload or read fails. Failures are + * non-fatal (they degrade to a miss), so without this hook an operator + * sees only the console-facing warning. Invoked for every failed + * operation; callers that log should dedupe if they want a single line. */ onUploadError?: (hash: string, error: unknown) => void; diff --git a/packages/tooling/task-runner/src/cache.ts b/packages/tooling/task-runner/src/cache.ts index c5138f1e..f7a9c240 100644 --- a/packages/tooling/task-runner/src/cache.ts +++ b/packages/tooling/task-runner/src/cache.ts @@ -53,6 +53,16 @@ interface CacheOptions { /** Maximum cache size (e.g., "500MB", "1GB") */ maxCacheSize?: string; + /** + * Called when the cache directory cannot be created (EACCES, EROFS, + * ENOSPC, …). The instance disables itself for the rest of the run: + * reads miss, writes are no-ops, and tasks execute normally. Setup + * therefore never crashes the run. The callback receives the directory + * that failed and the underlying error so the caller can name both in a + * warning. + */ + onSetupError?: (directory: string, error: unknown) => void; + /** * Diagnostic sink for best-effort cache writes that fail (ENOSPC, * EACCES, EROFS, …). {@link Cache.put} stays non-throwing so a @@ -190,12 +200,21 @@ class Cache { readonly #onWriteError: ((hash: string, error: unknown) => void) | undefined; + readonly #onSetupError: ((directory: string, error: unknown) => void) | undefined; + + /** Set when the cache directory could not be created; every operation degrades to a miss/no-op. */ + #disabled = false; + + /** Memoized setup so concurrent callers share one mkdir and one error report. */ + #readyPromise: Promise | undefined; + /** Serializes concurrent setTaskIndex writes to prevent lost updates */ #indexWriteQueue: Promise = Promise.resolve(); public constructor(options: CacheOptions) { this.#workspaceRoot = options.workspaceRoot; this.#onWriteError = options.onWriteError; + this.#onSetupError = options.onSetupError; const baseDirectory = options.cacheDirectory ?? join(options.workspaceRoot, DEFAULT_CACHE_DIRECTORY_NAME); @@ -214,6 +233,37 @@ class Cache { this.#maxCacheSize = options.maxCacheSize ? parseCacheSize(options.maxCacheSize) : undefined; } + /** + * Creates the cache directory once, before the run starts. Callers that + * route through this method get a clear degraded mode instead of an + * EACCES surfacing later from a per-task write or crashing setup. + * + * Kept asynchronous (rather than creating the directory in the + * constructor) so read-only users — cache inspection, `cache clean` — + * do not materialise a directory as a side effect of constructing Cache. + * Calling it more than once is safe. + */ + public ensureReady(): Promise { + this.#readyPromise ??= (async () => { + try { + await mkdir(this.#cacheDirectory, { recursive: true }); + } catch (error) { + this.#disabled = true; + this.#onSetupError?.(this.#cacheDirectory, error); + } + })(); + + return this.#readyPromise; + } + + /** + * True when the cache directory could not be created and the instance is + * running in degraded mode for this process. + */ + public get disabled(): boolean { + return this.#disabled; + } + /** * Gets the cache directory path. */ @@ -238,6 +288,10 @@ class Cache { * the workspace. */ public async getActionResult(actionDigest: CasDigest): Promise { + if (this.#disabled) { + return null; + } + return readActionEntry(this.#cacheDirectory, actionDigest.hash); } @@ -247,6 +301,10 @@ class Cache { * through to the legacy `<hash>/` layout (or executes the task). */ public async resolveActionDigestForTaskHash(taskHash: string): Promise { + if (this.#disabled) { + return null; + } + return readTaskHashIndex(this.#cacheDirectory, taskHash); } @@ -257,6 +315,10 @@ class Cache { * partial failure can't surface a half-written entry. */ public async putActionResult(taskHash: string, actionDigest: CasDigest, result: ActionResult, blobs: ReadonlyArray): Promise { + if (this.#disabled) { + return; + } + await this.#ensureBlobs(blobs); await writeActionEntry(this.#cacheDirectory, actionDigest.hash, result); await writeTaskHashIndex(this.#cacheDirectory, taskHash, actionDigest.hash); @@ -269,6 +331,10 @@ class Cache { * as a cache miss and re-executes. */ public async materializeOutputs(result: ActionResult, workspaceRoot: string): Promise { + if (this.#disabled) { + return false; + } + for (const file of result.outputFiles) { const destination = resolve(workspaceRoot, file.path); @@ -338,6 +404,10 @@ class Cache { * Returns undefined if no valid cache entry exists. */ public async get(hash: string): Promise { + if (this.#disabled) { + return undefined; + } + const cacheEntryDirectory = join(this.#cacheDirectory, hash); try { @@ -395,6 +465,10 @@ class Cache { fingerprint?: TaskFingerprint, autoWrites?: ReadonlyArray, ): Promise { + if (this.#disabled) { + return; + } + const cacheEntryDirectory = join(this.#cacheDirectory, hash); const temporaryDirectory = join(this.#cacheDirectory, `.tmp-${hash}-${uniqueId()}`); @@ -476,6 +550,10 @@ class Cache { * for backward compat. */ public async restoreOutputs(hash: string, _outputs?: OutputSpec[], options?: ExtractOptions): Promise { + if (this.#disabled) { + return false; + } + const cacheEntryDirectory = join(this.#cacheDirectory, hash); return restoreOutputsCompressed(cacheEntryDirectory, this.#workspaceRoot, options); @@ -487,6 +565,10 @@ class Cache { * tracked file accesses rather than computed upfront. */ public async getByTaskId(taskId: string): Promise { + if (this.#disabled) { + return undefined; + } + const indexFile = join(this.#cacheDirectory, ".task-index.json"); try { @@ -510,6 +592,10 @@ class Cache { * Each write is atomic (temp file + rename). */ public async setTaskIndex(taskId: string, hash: string): Promise { + if (this.#disabled) { + return; + } + // Serialize writes through a queue to prevent concurrent // read-modify-write races. Two promises here: one for THIS // caller (must reject if the write actually fails) and one @@ -523,7 +609,9 @@ class Cache { this.#indexWriteQueue = writePromise.catch(() => {}); - return writePromise; + // Await rather than return the promise so this async method has a + // single (void) return shape; rejection still reaches the caller. + await writePromise; } async #writeTaskIndex(taskId: string, hash: string): Promise { @@ -637,6 +725,10 @@ class Cache { */ public async removeOldEntries(): Promise { + if (this.#disabled) { + return; + } + try { const entries = await readdir(this.#cacheDirectory); const now = Date.now(); @@ -716,6 +808,10 @@ class Cache { * Clears the entire cache. */ public async clear(): Promise { + if (this.#disabled) { + return; + } + const directory = resolve(this.#cacheDirectory); // Never recursively delete the workspace root or the filesystem root — diff --git a/packages/tooling/task-runner/src/default-task-runner.ts b/packages/tooling/task-runner/src/default-task-runner.ts index f66cfe74..e83d78c7 100644 --- a/packages/tooling/task-runner/src/default-task-runner.ts +++ b/packages/tooling/task-runner/src/default-task-runner.ts @@ -265,18 +265,30 @@ const defaultTaskRunner = async (_tasks: Task[], options: TaskRunnerOptions, con ? computeGlobalEnvNamespace(options.globalEnv, (name) => childEnvVisible?.(name) ?? true) : undefined; - // Create the cache + // Create the cache. Readiness is established once here so an unusable + // directory (EACCES, EROFS, ENOSPC, …) becomes a reported degraded mode + // rather than a per-task failure. const cache = new Cache({ cacheDirectory: options.cacheDirectory, cacheNamespace, maxCacheAge: options.maxCacheAge, maxCacheSize: options.maxCacheSize, + onSetupError: (directory, error) => { + const message = error instanceof Error ? error.message : String(error); + + options.onDiagnostic?.( + "cache", + `Local cache disabled for this run: cannot create cache directory "${directory}" (${message}). Tasks will run without caching.`, + ); + }, onWriteError: options.onDiagnostic ? (hash, error) => options.onDiagnostic?.(hash, `cache write failed: ${error instanceof Error ? error.message : String(error)}`) : undefined, workspaceRoot, }); + await cache.ensureReady(); + // Clean old cache entries in the background cache.removeOldEntries().catch(() => {}); @@ -368,8 +380,41 @@ const defaultTaskRunner = async (_tasks: Task[], options: TaskRunnerOptions, con // Create remote cache if configured. `createRemoteCacheBackend` // centralizes backend selection so new backends (REAPI gRPC) can - // land without touching this file. - const remoteCache = options.remoteCache ? createRemoteCacheBackend(options.remoteCache) : undefined; + // land without touching this file. Remote caching is an optimisation: + // a transport failure on read or write degrades to a cache miss, and the + // operator sees one warning for the run rather than one per task. + let remoteCacheErrorReported = false; + const reportRemoteCacheError = (error: unknown): void => { + // The caller's hook still sees every failure, so debug logging can + // retain the full sequence; only the console-facing diagnostic is + // collapsed to one message per run. + options.remoteCache?.onUploadError?.("remote-cache", error); + + if (remoteCacheErrorReported) { + return; + } + + remoteCacheErrorReported = true; + + const message = error instanceof Error ? error.message : String(error); + + options.onDiagnostic?.("remote-cache", `Remote cache unavailable; continuing without it for this run: ${message}`); + }; + + const remoteCache = options.remoteCache + ? createRemoteCacheBackend( + { + ...options.remoteCache, + // HTTP downloads hydrate the local CAS before the cache entry + // is extracted. Default that root to this run's cache so a + // remote read can actually populate a local hit. + localCasRoot: options.remoteCache.localCasRoot ?? cache.cacheDirectory, + }, + (error) => { + reportRemoteCacheError(error); + }, + ) + : undefined; // Merge framework-inferred env patterns with explicitly configured ones // This makes frameworkInference work with both Nx-style and autoFingerprint modes @@ -395,9 +440,12 @@ const defaultTaskRunner = async (_tasks: Task[], options: TaskRunnerOptions, con fingerprintEnvPatterns, lifeCycle, // Bridge-local upload errors (tar/digest in hash-bridge) flow back - // through the same callback the wire-level backend uses, so a user - // observes upload failures regardless of where they originate. - onRemoteUploadError: options.remoteCache?.onUploadError, + // through the same once-per-run reporter the wire-level backend uses, + // so a user observes upload failures regardless of where they + // originate without the warning repeating for every task. + onRemoteUploadError: (_hash, error) => { + reportRemoteCacheError(error); + }, preTraceCheck: context.preTraceCheck, remoteCache, resolveChildEnv, diff --git a/packages/tooling/task-runner/src/external-inputs.ts b/packages/tooling/task-runner/src/external-inputs.ts new file mode 100644 index 00000000..03fb42d5 --- /dev/null +++ b/packages/tooling/task-runner/src/external-inputs.ts @@ -0,0 +1,288 @@ +// eslint-disable-next-line n/no-unsupported-features/node-builtins -- stable in the Node 22.x runtimes we ship against +import { glob, readFile, stat } from "node:fs/promises"; + +// eslint-disable-next-line import/no-extraneous-dependencies -- bundled inline by packem from workspace devDependency +import { xxh3Hash } from "@shared/xxh3"; +import { isAbsolute, relative, resolve } from "@visulima/path"; + +import { looksLikeInputUri, parseInputUri } from "./parse-input-uri"; +import type { FileSetInput, InputDefinition, NamedInputs, ProjectConfiguration, TargetConfiguration } from "./types"; +import { collectFiles } from "./utils"; + +/** + * A file-set input that resolves outside `workspaceRoot`, together with the + * projects whose targets declared it. + * + * The task hasher and affected detection must agree on this set: the hasher + * folds the external files into the cache key, while affected detection uses + * the same owner map to decide which projects an external change touches. + */ +export interface ExternalInputDefinition { + /** Absolute glob or literal path, with forward slashes. */ + pattern: string; + /** Projects that declared this external input. */ + projects: string[]; +} + +/** Directory names the native directory walker skips, mirrored here. */ +const IGNORED_DIRECTORIES = new Set([".cache", ".git", "coverage", "dist", "node_modules"]); + +const GLOB_METACHARS = /[*?[\]{}]/; + +/** + * Normalizes a fileset entry into a string pattern that uses + * `{projectRoot}` / `{workspaceRoot}` tokens. + * + * - Bare strings are returned unchanged. + * - Object form `{ pattern, base: "workspace" }` → `{workspaceRoot}/<pattern>` + * (or `!{workspaceRoot}/...` when `pattern` starts with `!`). + * - Object form `{ pattern, base: "package" }` → `{projectRoot}/<pattern>`. + */ +export const normalizeFileset = (fileset: FileSetInput["fileset"]): string => { + if (typeof fileset === "string") { + return fileset; + } + + const token = fileset.base === "workspace" ? "{workspaceRoot}" : "{projectRoot}"; + + if (fileset.pattern.startsWith("!")) { + return `!${token}/${fileset.pattern.slice(1)}`; + } + + return `${token}/${fileset.pattern}`; +}; + +/** + * Resolves a fileset pattern against the workspace. Returns an absolute, + * forward-slash pattern with a leading `!` preserved for negations. + */ +export const resolveFilesetPattern = (workspaceRoot: string, projectRoot: string, fileset: FileSetInput["fileset"]): string => { + const normalized = normalizeFileset(fileset); + const negated = normalized.startsWith("!"); + const body = negated ? normalized.slice(1) : normalized; + const resolved = resolve(workspaceRoot, body.replaceAll("{projectRoot}", () => projectRoot).replaceAll("{workspaceRoot}", () => ".")); + + return negated ? `!${resolved}` : resolved; +}; + +/** + * The deepest directory prefix of a pattern that contains no glob magic. + * Used to decide containment without treating a wildcard as a path segment. + */ +const nonGlobBase = (pattern: string): string => { + const segments = pattern.split("/"); + const firstGlobIndex = segments.findIndex((segment) => GLOB_METACHARS.test(segment)); + + return (firstGlobIndex === -1 ? segments : segments.slice(0, firstGlobIndex)).join("/") || "/"; +}; + +/** + * Returns `true` when a resolved pattern addresses something outside the + * workspace root. Glob magic is removed before the containment check so a + * pattern like `/external/**` is evaluated on `/external`. + */ +export const isExternalPattern = (workspaceRoot: string, pattern: string): boolean => { + const absolute = pattern.startsWith("!") ? pattern.slice(1) : pattern; + const rel = relative(workspaceRoot, nonGlobBase(absolute)); + + return rel === ".." || rel.startsWith("../") || isAbsolute(rel); +}; + +const isFileSetInput = (input: InputDefinition): input is FileSetInput => "fileset" in input; + +/** Expands named inputs and URI sugar into the raw fileset patterns they carry. */ +const collectFilesetPatterns = ( + inputs: (string | InputDefinition)[], + namedInputs: NamedInputs, + seen: Set, +): FileSetInput["fileset"][] => { + const patterns: FileSetInput["fileset"][] = []; + + for (const input of inputs) { + if (typeof input === "string") { + if (looksLikeInputUri(input)) { + const parsed = parseInputUri(input); + + if (parsed && isFileSetInput(parsed)) { + patterns.push(parsed.fileset); + } + + continue; + } + + if (input.startsWith("{") || input.startsWith("!{")) { + patterns.push(input); + } else if (input.startsWith("^")) { + continue; + } else if (namedInputs[input] && !seen.has(input)) { + seen.add(input); + patterns.push(...collectFilesetPatterns(namedInputs[input], namedInputs, seen)); + } else { + patterns.push(input); + } + } else if (isFileSetInput(input)) { + patterns.push(input.fileset); + } + } + + return patterns; +}; + +/** + * Collects every file-set input declared by a project target that resolves + * outside `workspaceRoot`, grouped by the projects that declare it. + * + * Both the declared hasher (for cache keys) and affected detection (for + * project selection) build on this function, so they cannot disagree about + * which external paths belong to which project. + */ +export const collectExternalInputs = ( + workspaceRoot: string, + projects: Record, + options: { namedInputs?: NamedInputs; targetDefaults?: Record> } = {}, +): ExternalInputDefinition[] => { + const namedInputs = options.namedInputs ?? {}; + const targetDefaults = options.targetDefaults ?? {}; + const owners = new Map>(); + + for (const [projectName, project] of Object.entries(projects)) { + const projectRoot = project.root ?? ""; + const targets = new Map>(Object.entries(targetDefaults)); + + for (const [targetName, target] of Object.entries(project.targets ?? {})) { + targets.set(targetName, { ...targets.get(targetName), ...target }); + } + + for (const target of targets.values()) { + if (!target.inputs) { + continue; + } + + for (const fileset of collectFilesetPatterns(target.inputs, namedInputs, new Set())) { + const pattern = resolveFilesetPattern(workspaceRoot, projectRoot, fileset); + + if (pattern.startsWith("!") || !isExternalPattern(workspaceRoot, pattern)) { + continue; + } + + let projectNames = owners.get(pattern); + + if (!projectNames) { + projectNames = new Set(); + owners.set(pattern, projectNames); + } + + projectNames.add(projectName); + } + } + } + + return [...owners.entries()] + .map(([pattern, projectNames]) => { return { pattern, projects: [...projectNames].toSorted() }; }) + .toSorted((a, b) => a.pattern.localeCompare(b.pattern)); +}; + +/** Expands a literal path or glob into the sorted files it addresses. */ +const expandPattern = async (pattern: string): Promise => { + if (!GLOB_METACHARS.test(pattern)) { + const direct = await stat(pattern).catch(() => undefined); + + if (direct?.isFile()) { + return [pattern]; + } + + if (direct?.isDirectory()) { + const files = await collectFiles(pattern, IGNORED_DIRECTORIES); + + return files.toSorted(); + } + + return []; + } + + try { + const entries: string[] = []; + + for await (const entry of glob(pattern, { withFileTypes: false })) { + if (typeof entry === "string" && !entry.split(/[/\\]/).some((segment) => IGNORED_DIRECTORIES.has(segment))) { + entries.push(entry); + } + } + + return entries.toSorted(); + } catch { + return []; + } +}; + +/** + * Content-hashes every file addressed by the external patterns. The result + * is keyed by the pattern itself, so callers can compare a previous snapshot + * with the current state and find the patterns (and therefore projects) whose + * external inputs changed. + * + * A pattern that currently matches nothing hashes as an empty string rather + * than being omitted: a file appearing later is then a change. + */ +export const computeExternalInputHashes = async (inputs: ReadonlyArray): Promise> => { + const hashes: Record = {}; + + for (const input of inputs) { + // eslint-disable-next-line no-await-in-loop -- patterns are hashed sequentially to keep IO pressure predictable + const files = await expandPattern(input.pattern); + const parts: string[] = []; + + for (const file of files) { + try { + // eslint-disable-next-line no-await-in-loop -- per-file reads are bounded by the pattern's file count + const content = await readFile(file); + + parts.push(`${file}:${xxh3Hash(content)}`); + } catch { + // Unreadable files contribute a stable marker so gaining or + // losing read permission still changes the snapshot. + parts.push(`${file}:__unreadable__`); + } + } + + hashes[input.pattern] = xxh3Hash(Buffer.from(parts.join("\n"), "utf8")); + } + + return hashes; +}; + +/** + * Returns the subset of external patterns whose content differs from the + * previous snapshot. When no previous snapshot exists, every pattern counts + * as changed — affected detection must not report "nothing changed" for + * external inputs it has never observed. + */ +export const findChangedExternalInputs = ( + inputs: ReadonlyArray, + current: Record, + previous: Record | undefined, +): string[] => + inputs + .filter((input) => previous === undefined || previous[input.pattern] !== current[input.pattern]) + .map((input) => input.pattern); + +/** Returns the projects that own at least one of the given external patterns. */ +export const projectsForExternalPatterns = ( + inputs: ReadonlyArray, + patterns: ReadonlyArray, +): string[] => { + const changed = new Set(patterns); + const projects = new Set(); + + for (const input of inputs) { + if (!changed.has(input.pattern)) { + continue; + } + + for (const project of input.projects) { + projects.add(project); + } + } + + return [...projects]; +}; diff --git a/packages/tooling/task-runner/src/index.ts b/packages/tooling/task-runner/src/index.ts index 02df17eb..36864ca1 100644 --- a/packages/tooling/task-runner/src/index.ts +++ b/packages/tooling/task-runner/src/index.ts @@ -58,6 +58,17 @@ export { runConcurrentFallback } from "./concurrent-fallback"; export { DEFAULT_UNTRACKED_ENV_PATTERNS, defaultTaskRunner } from "./default-task-runner"; // Shell detection export { detectScriptShell } from "./detect-shell"; +// External file-set inputs (outside the workspace root) +export type { ExternalInputDefinition } from "./external-inputs"; +export { + collectExternalInputs, + computeExternalInputHashes, + findChangedExternalInputs, + isExternalPattern, + normalizeFileset, + projectsForExternalPatterns, + resolveFilesetPattern, +} from "./external-inputs"; // File access tracking export type { FileAccess, TrackingResult } from "./file-access-tracker"; export { FileAccessTracker, generatePreloadScript } from "./file-access-tracker"; diff --git a/packages/tooling/task-runner/src/task-hasher.ts b/packages/tooling/task-runner/src/task-hasher.ts index f003931f..3f89e7ea 100644 --- a/packages/tooling/task-runner/src/task-hasher.ts +++ b/packages/tooling/task-runner/src/task-hasher.ts @@ -5,8 +5,9 @@ import { matchesGlob } from "node:path"; // eslint-disable-next-line import/no-extraneous-dependencies -- bundled inline by packem from workspace devDependency import { createXxh3Hasher, xxh3Hash } from "@shared/xxh3"; -import { join, resolve } from "@visulima/path"; +import { isAbsolute, join, relative, resolve } from "@visulima/path"; +import { normalizeFileset, resolveFilesetPattern } from "./external-inputs"; import { createEnvMatcher, resolveEnvPatterns } from "./fingerprint"; import { getFrameworkEnvVariables } from "./framework-inference"; import type { IncrementalFileHasher } from "./incremental-hasher"; @@ -290,29 +291,6 @@ export const extractReferencedEnvVars = (command: string): string[] => { // Type guards const isFileSetInput = (input: InputDefinition): input is FileSetInput => "fileset" in input; -/** - * Normalizes a fileset entry into a string pattern that uses - * `{projectRoot}` / `{workspaceRoot}` tokens for downstream resolution. - * - * - Bare strings are returned unchanged. - * - Object form `{ pattern, base: "workspace" }` → `{workspaceRoot}/<pattern>` - * (or `!{workspaceRoot}/...` when `pattern` starts with `!`). - * - Object form `{ pattern, base: "package" }` → `{projectRoot}/<pattern>`. - */ -const normalizeFileset = (fileset: FileSetInput["fileset"]): string => { - if (typeof fileset === "string") { - return fileset; - } - - const token = fileset.base === "workspace" ? "{workspaceRoot}" : "{projectRoot}"; - - if (fileset.pattern.startsWith("!")) { - return `!${token}/${fileset.pattern.slice(1)}`; - } - - return `${token}/${fileset.pattern}`; -}; - const isRuntimeInput = (input: InputDefinition): input is RuntimeInput => "runtime" in input; const isEnvironmentInput = (input: InputDefinition): input is EnvironmentInput => "env" in input; @@ -664,7 +642,7 @@ class InProcessTaskHasher implements TaskHasher { continue; } - const resolved = normalizeFileset(input.fileset).replace("{projectRoot}", projectRoot).replace("{workspaceRoot}", "."); + const resolved = resolveFilesetPattern(this.#workspaceRoot, projectRoot, input.fileset); if (resolved.startsWith("!")) { patterns.push(resolved.slice(1)); @@ -682,7 +660,7 @@ class InProcessTaskHasher implements TaskHasher { const project = this.#projects[task.target.project]; const projectRoot = project?.root ?? ""; - const resolvedPattern = pattern.replace("{projectRoot}", projectRoot).replace("{workspaceRoot}", "."); + const resolvedPattern = resolveFilesetPattern(this.#workspaceRoot, projectRoot, pattern); if (resolvedPattern.startsWith("!")) { // Negation patterns are collected and applied by the caller @@ -733,7 +711,13 @@ class InProcessTaskHasher implements TaskHasher { continue; } - result[path] = hash; + // Key by the path relative to the workspace, not the native + // walker's spelling. Inside the root the walker already returns + // the relative path, so only the rarer external case pays for + // the extra path arithmetic. Outside it, this keeps the key + // deterministic across machines instead of embedding an + // absolute checkout path in the cache key. + result[isAbsolute(path) ? relative(this.#workspaceRoot, path) : path] = hash; // Intentionally NOT seeded into `#fileHashCache`: that cache is // mtime/size-revalidated in `#hashFile`, but the native batch diff --git a/packages/tooling/task-runner/src/task-orchestrator.ts b/packages/tooling/task-runner/src/task-orchestrator.ts index 82d3bce3..d823c306 100644 --- a/packages/tooling/task-runner/src/task-orchestrator.ts +++ b/packages/tooling/task-runner/src/task-orchestrator.ts @@ -646,7 +646,7 @@ class TaskOrchestrator { } if (this.#remoteCache) { - const retrieved = await retrieveByTaskHash(this.#remoteCache, hash, this.#cache.cacheDirectory); + const retrieved = await retrieveByTaskHash(this.#remoteCache, hash, this.#cache.cacheDirectory, this.#onRemoteUploadError); if (retrieved) { const remoteCached = await this.#cache.get(hash); diff --git a/packages/tooling/vis/__tests__/commands/run/run-remote-cache.test.ts b/packages/tooling/vis/__tests__/commands/run/run-remote-cache.test.ts new file mode 100644 index 00000000..4c9a9c11 --- /dev/null +++ b/packages/tooling/vis/__tests__/commands/run/run-remote-cache.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it, vi } from "vitest"; + +import { applyRemoteCacheConfig } from "../../../src/commands/run/run-remote-cache"; + +describe(applyRemoteCacheConfig, () => { + it("logs every remote cache failure at debug level and preserves a configured hook", async () => { + expect.assertions(3); + + const configuredOnUploadError = vi.fn<(hash: string, error: unknown) => void>(); + const logger = { + debug: vi.fn<(message: string) => void>(), + info: vi.fn<(message: string) => void>(), + warn: vi.fn<(message: string) => void>(), + }; + const runnerOptions = {} as Parameters[0]["runnerOptions"]; + + await applyRemoteCacheConfig({ + configTaskRunner: { + remoteCache: { + onUploadError: configuredOnUploadError, + url: "https://cache.example.com", + }, + }, + logger: logger as never, + options: {} as never, + runnerOptions, + workspaceRoot: "/tmp/workspace", + }); + + runnerOptions.remoteCache?.onUploadError?.("hash-1", new Error("connect ECONNREFUSED")); + runnerOptions.remoteCache?.onUploadError?.("hash-2", "plain failure"); + + expect(configuredOnUploadError).toHaveBeenCalledTimes(2); + expect(logger.debug).toHaveBeenNthCalledWith(1, expect.stringContaining("connect ECONNREFUSED")); + expect(logger.debug).toHaveBeenNthCalledWith(2, expect.stringContaining("plain failure")); + }); +}); diff --git a/packages/tooling/vis/__tests__/commands/run/run-teardown-on-throw.test.ts b/packages/tooling/vis/__tests__/commands/run/run-teardown-on-throw.test.ts index 38642781..d6c4f9df 100644 --- a/packages/tooling/vis/__tests__/commands/run/run-teardown-on-throw.test.ts +++ b/packages/tooling/vis/__tests__/commands/run/run-teardown-on-throw.test.ts @@ -50,7 +50,7 @@ vi.mock(import("../../../src/runtime/command-runtime"), () => { }); vi.mock(import("../../../src/util/vis-paths"), () => { - return { cleanupLegacyTaskRunnerLayout: () => {} }; + return { cleanupLegacyTaskRunnerLayout: () => {}, DEFAULT_WORKSPACE_CACHE_DIRECTORY: "node_modules/.cache/vis" }; }); vi.mock(import("../../../src/config/workspace"), () => { diff --git a/packages/tooling/vis/__tests__/task/affected-selection.test.ts b/packages/tooling/vis/__tests__/task/affected-selection.test.ts index 11f9e9a6..334482d3 100644 --- a/packages/tooling/vis/__tests__/task/affected-selection.test.ts +++ b/packages/tooling/vis/__tests__/task/affected-selection.test.ts @@ -1,8 +1,12 @@ +import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + import type { ProjectConfiguration, ProjectGraph } from "@visulima/task-runner"; import { describe, expect, it, vi } from "vitest"; import { VisUserError } from "../../src/errors/vis-user-error"; -import { parsePorcelainStatus, selectAffectedProjects } from "../../src/task/affected-selection"; +import { invalidateExternalInputState, parsePorcelainStatus, persistExternalInputHashes, selectAffectedProjects } from "../../src/task/affected-selection"; const { getAffectedProjectsMock } = vi.hoisted(() => { return { getAffectedProjectsMock: vi.fn() }; @@ -216,3 +220,203 @@ describe(selectAffectedProjects, () => { expect(result.notes).toStrictEqual(["ignoring 2 changed path(s) outside any project matching affectedIgnore"]); }); }); + +describe("selectAffectedProjects external inputs", () => { + it("defaults the external-input snapshot to the workspace cache directory", async () => { + expect.assertions(1); + + const root = await mkdtemp(join(tmpdir(), "vis-affected-external-default-")); + const externalDirectory = await mkdtemp(join(tmpdir(), "vis-affected-shared-default-")); + const externalFile = join(externalDirectory, "shared.ts"); + + await mkdir(join(root, "packages/api"), { recursive: true }); + await writeFile(externalFile, "export const shared = 1;\n"); + + const externalProjects: Record = { + api: { root: "packages/api", targets: { build: { inputs: [`file://${externalFile}`] } } }, + }; + + try { + getAffectedProjectsMock.mockResolvedValueOnce({ + ...emptyResult, + changedExternalInputs: [externalFile], + changedProjects: ["api"], + externalInputHashes: { [externalFile]: "hash-default" }, + }); + + await selectAffectedProjects( + { base: "HEAD", head: "HEAD", uncommitted: false }, + { + projectGraph, + projects: externalProjects, + workspaceRoot: root, + }, + ); + + const snapshotPath = join(root, "node_modules", ".cache", "vis", "external-inputs.json"); + + expect(JSON.parse(await readFile(snapshotPath, "utf8"))).toStrictEqual({ [externalFile]: "hash-default" }); + } finally { + await rm(root, { force: true, recursive: true }); + await rm(externalDirectory, { force: true, recursive: true }); + } + }); + + it("does not advance the snapshot when persistence is deferred", async () => { + expect.assertions(1); + + const root = await mkdtemp(join(tmpdir(), "vis-affected-external-deferred-")); + const externalDirectory = await mkdtemp(join(tmpdir(), "vis-affected-shared-deferred-")); + const externalFile = join(externalDirectory, "shared.ts"); + const statePath = join(root, "external-inputs.json"); + + await mkdir(join(root, "packages/api"), { recursive: true }); + await writeFile(externalFile, "export const shared = 1;\n"); + + const externalProjects: Record = { + api: { root: "packages/api", targets: { build: { inputs: [`file://${externalFile}`] } } }, + }; + + try { + getAffectedProjectsMock.mockResolvedValueOnce({ + ...emptyResult, + changedExternalInputs: [externalFile], + changedProjects: ["api"], + externalInputHashes: { [externalFile]: "hash-deferred" }, + }); + + await selectAffectedProjects( + { base: "HEAD", head: "HEAD", uncommitted: false }, + { projectGraph, projects: externalProjects, workspaceRoot: root }, + { externalInputStatePath: statePath, persistExternalInputState: false }, + ); + + await expect(readFile(statePath, "utf8")).rejects.toThrow(/ENOENT/); + } finally { + await rm(root, { force: true, recursive: true }); + await rm(externalDirectory, { force: true, recursive: true }); + } + }); + + it("ignores a snapshot that is not a plain object", async () => { + expect.assertions(1); + + const root = await mkdtemp(join(tmpdir(), "vis-affected-external-shape-")); + const externalDirectory = await mkdtemp(join(tmpdir(), "vis-affected-shared-shape-")); + const externalFile = join(externalDirectory, "shared.ts"); + const statePath = join(root, "external-inputs.json"); + + await mkdir(join(root, "packages/api"), { recursive: true }); + await writeFile(externalFile, "export const shared = 1;\n"); + await writeFile(statePath, "null"); + + const externalProjects: Record = { + api: { root: "packages/api", targets: { build: { inputs: [`file://${externalFile}`] } } }, + }; + + try { + getAffectedProjectsMock.mockResolvedValueOnce({ ...emptyResult, externalInputHashes: {} }); + + await selectAffectedProjects( + { base: "HEAD", head: "HEAD", uncommitted: false }, + { projectGraph, projects: externalProjects, workspaceRoot: root }, + { externalInputStatePath: statePath, persistExternalInputState: false }, + ); + + const call = getAffectedProjectsMock.mock.calls.at(-1)?.[0]; + + expect(call.previousExternalInputHashes).toBeUndefined(); + } finally { + await rm(root, { force: true, recursive: true }); + await rm(externalDirectory, { force: true, recursive: true }); + } + }); + + it("passes the shared external-input ownership map and persists its hashes", async () => { + expect.assertions(6); + + const root = await mkdtemp(join(tmpdir(), "vis-affected-external-")); + const externalDirectory = await mkdtemp(join(tmpdir(), "vis-affected-shared-")); + const externalFile = join(externalDirectory, "shared.ts"); + const statePath = join(root, ".vis", "external-inputs.json"); + + await mkdir(join(root, "packages/api"), { recursive: true }); + await writeFile(externalFile, "export const shared = 1;\n"); + + const externalProjects: Record = { + api: { root: "packages/api", targets: { build: { inputs: [`file://${externalFile}`] } } }, + }; + + try { + getAffectedProjectsMock.mockResolvedValueOnce({ + ...emptyResult, + changedExternalInputs: [externalFile], + changedProjects: ["api"], + externalInputHashes: { [externalFile]: "hash-1" }, + }); + + const first = await selectAffectedProjects( + { base: "HEAD", head: "HEAD", uncommitted: false }, + { + projectGraph, + projects: externalProjects, + workspaceRoot: root, + }, + { externalInputStatePath: statePath }, + ); + + const firstCall = getAffectedProjectsMock.mock.calls.at(-1)?.[0]; + + expect(firstCall.externalInputs).toStrictEqual([{ pattern: externalFile, projects: ["api"] }]); + expect(firstCall.previousExternalInputHashes).toBeUndefined(); + expect(first.changedProjects).toStrictEqual(["api"]); + expect(JSON.parse(await readFile(statePath, "utf8"))).toStrictEqual({ [externalFile]: "hash-1" }); + + getAffectedProjectsMock.mockResolvedValueOnce({ + ...emptyResult, + changedExternalInputs: [], + externalInputHashes: { [externalFile]: "hash-1" }, + }); + + await selectAffectedProjects( + { base: "HEAD", head: "HEAD", uncommitted: false }, + { + projectGraph, + projects: externalProjects, + workspaceRoot: root, + }, + { externalInputStatePath: statePath }, + ); + + const secondCall = getAffectedProjectsMock.mock.calls.at(-1)?.[0]; + + expect(secondCall.previousExternalInputHashes).toStrictEqual({ [externalFile]: "hash-1" }); + expect(first.notes.join("\n")).toMatch(/changed external input/); + } finally { + await rm(root, { force: true, recursive: true }); + await rm(externalDirectory, { force: true, recursive: true }); + } + }); +}); + +describe("external input snapshot persistence", () => { + it("writes the snapshot atomically and can invalidate it after a failure", async () => { + expect.assertions(3); + + const root = await mkdtemp(join(tmpdir(), "vis-affected-snapshot-")); + const statePath = join(root, "external-inputs.json"); + + try { + await persistExternalInputHashes(statePath, { pattern: "hash-1" }); + + expect(JSON.parse(await readFile(statePath, "utf8"))).toStrictEqual({ pattern: "hash-1" }); + await expect(readFile(`${statePath}.${process.pid}.tmp`, "utf8")).rejects.toThrow(/ENOENT/); + + await invalidateExternalInputState(statePath); + + await expect(readFile(statePath, "utf8")).rejects.toThrow(/ENOENT/); + } finally { + await rm(root, { force: true, recursive: true }); + } + }); +}); diff --git a/packages/tooling/vis/src/cache/cache-directory.ts b/packages/tooling/vis/src/cache/cache-directory.ts index 9492c670..11b1067a 100644 --- a/packages/tooling/vis/src/cache/cache-directory.ts +++ b/packages/tooling/vis/src/cache/cache-directory.ts @@ -188,6 +188,27 @@ export const applyBranchScope = (cacheDirectory: string, workspaceRoot: string, return resolve(cacheDirectory, "branches", slug); }; +/** + * Resolves the cache directory a `vis run` invocation will use: the shared + * worktree anchor, explicit path precedence, and branch scoping applied in one + * place. Both the runner and affected detection call this so the external-input + * snapshot lives beside the cache entries it describes. + */ +export const resolveRunCacheDirectory = ({ + branchScopedCache, + configCacheDir, + optionsCacheDir, + sharedWorktreeCache, + workspaceRoot, +}: { + branchScopedCache: boolean | undefined; + configCacheDir: string | undefined; + optionsCacheDir: string | undefined; + sharedWorktreeCache: boolean | undefined; + workspaceRoot: string; +}): string => + applyBranchScope(resolveSharedCacheDirectory(workspaceRoot, optionsCacheDir, configCacheDir, sharedWorktreeCache), workspaceRoot, branchScopedCache); + /** * Returns `true` when `cacheDirectory` sits inside `workspaceRoot`. Uses * `relative()` instead of `startsWith()` so paths with a common string diff --git a/packages/tooling/vis/src/commands/affected/handler.ts b/packages/tooling/vis/src/commands/affected/handler.ts index b25c6bfe..b35edff3 100644 --- a/packages/tooling/vis/src/commands/affected/handler.ts +++ b/packages/tooling/vis/src/commands/affected/handler.ts @@ -1,5 +1,7 @@ import type { CommandExecute, Toolbox } from "@visulima/cerebro"; +import { join } from "@visulima/path"; +import { resolveRunCacheDirectory } from "../../cache/cache-directory"; import { buildProjectGraph, discoverWorkspace } from "../../config/workspace"; import { VisUserError } from "../../errors/vis-user-error"; import { selectAffectedProjects } from "../../task/affected-selection"; @@ -68,8 +70,16 @@ const execute = async ({ // over rather than maintaining a second copy of the filter pipeline. if (options.sparseCheckout) { const workspaceRoot = wsRoot; - const { fileOwners, packageJsons, workspace } = discoverWorkspace(workspaceRoot, visConfig); + const { config, fileOwners, packageJsons, workspace } = discoverWorkspace(workspaceRoot, visConfig); const projectGraph = buildProjectGraph(workspaceRoot, workspace, packageJsons); + const configTaskRunner = (config.taskRunner ?? {}) as { cacheDirectory?: string }; + const cacheDirectory = resolveRunCacheDirectory({ + branchScopedCache: config.branchScopedCache, + configCacheDir: configTaskRunner.cacheDirectory, + optionsCacheDir: undefined, + sharedWorktreeCache: config.sharedWorktreeCache, + workspaceRoot, + }); const result = await selectAffectedProjects( { @@ -79,8 +89,12 @@ const execute = async ({ uncommitted: options.uncommitted, upstream: options.upstream, }, - { fileOwners, projectGraph, projects: workspace.projects, workspaceRoot }, - { affectedIgnore: visConfig?.affectedIgnore, defaultBase: visConfig?.defaultBase }, + { fileOwners, namedInputs: visConfig?.namedInputs, projectGraph, projects: workspace.projects, workspaceRoot }, + { + affectedIgnore: visConfig?.affectedIgnore, + defaultBase: visConfig?.defaultBase, + externalInputStatePath: join(cacheDirectory, "external-inputs.json"), + }, ); let affectedProjects = filterProjectsByQuery(result.affectedProjects, workspace, options.query); diff --git a/packages/tooling/vis/src/commands/run/handler.ts b/packages/tooling/vis/src/commands/run/handler.ts index 0407f936..78c05168 100644 --- a/packages/tooling/vis/src/commands/run/handler.ts +++ b/packages/tooling/vis/src/commands/run/handler.ts @@ -2,8 +2,10 @@ import type { CommandExecute, Toolbox } from "@visulima/cerebro"; import type { ProcessEvent, TaskResults } from "@visulima/task-runner"; import { runConcurrently, TerminalBuffer } from "@visulima/task-runner"; +import { resolveRunCacheDirectory } from "../../cache/cache-directory"; import { buildProjectGraph, discoverWorkspace, loadVisTaskConfigsForWorkspace } from "../../config/workspace"; import { resolveCommandRuntime } from "../../runtime/command-runtime"; +import { invalidateExternalInputState, persistExternalInputHashes } from "../../task/affected-selection"; import { cleanupLegacyTaskRunnerLayout } from "../../util/vis-paths"; import type { RunOptions } from "./index"; import { MAX_OUTPUT_BYTES } from "./output"; @@ -62,9 +64,18 @@ const execute = async ({ argument, logger, options, visConfig, workspaceRoot: ws const taskConfigs = await loadVisTaskConfigsForWorkspace(workspaceRoot, visConfig); const { config, fileOwners, packageJsons, projectOptions, workspace } = discoverWorkspace(workspaceRoot, visConfig, taskConfigs); const projectGraph = buildProjectGraph(workspaceRoot, workspace, packageJsons); + const configTaskRunner = (config.taskRunner ?? {}) as { cacheDirectory?: string }; + const runCacheDirectory = resolveRunCacheDirectory({ + branchScopedCache: config.branchScopedCache, + configCacheDir: configTaskRunner.cacheDirectory, + optionsCacheDir: options.cacheDir, + sharedWorktreeCache: config.sharedWorktreeCache, + workspaceRoot, + }); const selection = await resolveRunSelection({ argument, + cacheDirectory: runCacheDirectory, config, fileOwners, logger, @@ -81,7 +92,7 @@ const execute = async ({ argument, logger, options, visConfig, workspaceRoot: ws return; } - const { affectedFiles, currentOs, forwardedArgs, projectsWithTarget, projectTargetIndex, target } = selection; + const { affectedExternalInputState, affectedFiles, currentOs, forwardedArgs, projectsWithTarget, projectTargetIndex, target } = selection; const runTasks = buildRunTasks({ affectedFiles, @@ -159,6 +170,7 @@ const execute = async ({ argument, logger, options, visConfig, workspaceRoot: ws options, persistentTasks, projectsWithTarget, + runCacheDirectory, serviceEnvByTaskId, serviceResult, target, @@ -167,6 +179,27 @@ const execute = async ({ argument, logger, options, visConfig, workspaceRoot: ws workspaceRoot, }); + if (affectedExternalInputState) { + runtime.hooks.hook("run:after", async (results) => { + const failed = [...results.values()].some((result) => result.code !== 0); + + if (!failed) { + await persistExternalInputHashes(affectedExternalInputState.path, affectedExternalInputState.hashes); + } + }); + + // Persistent tasks start after `run:after` and never appear in the + // results map. If one exits non-zero, drop the snapshot written + // above so the next selection re-checks those external inputs. + const persistentTaskIds = new Set(persistentTasks.map((task) => task.id)); + + runtime.hooks.hook("task:failure", async (task) => { + if (persistentTaskIds.has(task.id)) { + await invalidateExternalInputState(affectedExternalInputState.path); + } + }); + } + await runtime.hooks.callHook("run:before", { tasks: initialTasks, workspaceRoot }); if (runtime.isTTY) { diff --git a/packages/tooling/vis/src/commands/run/run-remote-cache.ts b/packages/tooling/vis/src/commands/run/run-remote-cache.ts index 1279cde3..3ba76a4f 100644 --- a/packages/tooling/vis/src/commands/run/run-remote-cache.ts +++ b/packages/tooling/vis/src/commands/run/run-remote-cache.ts @@ -41,6 +41,17 @@ export const applyRemoteCacheConfig = async ({ ...(cliBackend ? { backend: cliBackend } : {}), }; + // Remote cache failures degrade to misses and produce one console + // warning per run. Keep the full sequence available under --debug so + // repeated connection or upload errors can still be diagnosed. Any + // hook supplied programmatically keeps running first. + const configuredOnUploadError = runnerOptions.remoteCache.onUploadError; + + runnerOptions.remoteCache.onUploadError = (hash, error) => { + configuredOnUploadError?.(hash, error); + logger.debug?.(`[vis run] remote cache error (${hash}): ${error instanceof Error ? error.message : String(error)}`); + }; + // Config can only carry serializable attestation knobs // (`requireOnDownload`, `expectedIdentity`); the actual keyless // sign/verify callbacks can't live in vis.config.ts. When the diff --git a/packages/tooling/vis/src/commands/run/run-runtime.ts b/packages/tooling/vis/src/commands/run/run-runtime.ts index b0b53a84..15df1a4f 100644 --- a/packages/tooling/vis/src/commands/run/run-runtime.ts +++ b/packages/tooling/vis/src/commands/run/run-runtime.ts @@ -3,7 +3,6 @@ import type { ChildEnvOptions, Task, TaskGraph, TaskResults, TaskRunnerOptions } import { generateRunSummary, pickChildParentEnv, writeChromeTrace } from "@visulima/task-runner"; import isInCi from "is-in-ci"; -import { applyBranchScope, resolveSharedCacheDirectory } from "../../cache/cache-directory"; import { FailureLogLifeCycle } from "../../report/failure-log"; import { checkStrictEnv, formatStrictEnvError } from "../../task/strict-env"; import type { detectCurrentOs } from "../../task/target-options"; @@ -36,6 +35,7 @@ export const createRunRuntime = async ({ options, persistentTasks, projectsWithTarget, + runCacheDirectory, serviceEnvByTaskId, serviceResult, target, @@ -52,6 +52,7 @@ export const createRunRuntime = async ({ options: RunOptions; persistentTasks: Task[]; projectsWithTarget: string[]; + runCacheDirectory: string; serviceEnvByTaskId: ApplyServiceRegistryResult["serviceEnvByTaskId"]; serviceResult: ApplyServiceRegistryResult; target: string; @@ -95,20 +96,11 @@ export const createRunRuntime = async ({ logger.info(`Profile written to ${profilePath}`); }; - // Resolve the cache directory through the shared helper so precedence - // (CLI > vis.config.ts > default) stays consistent with `vis cache` - // and relative --cache-dir values are normalized against the - // workspace root. Other fields keep their existing spread semantics - // to avoid changing override precedence for parallel/dryRun/etc. + // The handler resolves the cache directory once so affected selection and + // the runner write external-input state and cache entries to the same + // location. Other fields keep their existing spread semantics to avoid + // changing override precedence for parallel/dryRun/etc. const configTaskRunner = (config.taskRunner ?? {}) as TaskRunnerOptions & { cacheDirectory?: string }; - // Anchor the default cache path to the *main* worktree root when this - // workspace is a linked git worktree, so sibling agents share one cache - // instead of rebuilding the same hash N times. Explicit paths - // (--cache-dir, vis.config.ts, VIS_CACHE_DIRECTORY) win unchanged. - const baseCacheDirectory = resolveSharedCacheDirectory(workspaceRoot, options.cacheDir, configTaskRunner.cacheDirectory, config.sharedWorktreeCache); - // Branch-scope the cache dir when configured so main/feature - // branches stop overwriting each other's entries. - const resolvedCacheDirectory = applyBranchScope(baseCacheDirectory, workspaceRoot, config.branchScopedCache); // Concurrency resolution order: --parallel CLI flag > VIS_RUN_CONCURRENCY_LIMIT // env > built-in default of 3. The env variable lets CI runners set a @@ -142,7 +134,7 @@ export const createRunRuntime = async ({ // Applied after the config spread so the user's `--cache-dir` flag // wins over a config value and relative paths are normalized // against `workspaceRoot` via `resolveCacheDirectory()`. - cacheDirectory: resolvedCacheDirectory, + cacheDirectory: runCacheDirectory, // Forward the vis-owned data directory so run summaries land in // `/.vis/` rather than the task-runner default // `/.task-runner/`. The config-supplied value diff --git a/packages/tooling/vis/src/commands/run/run-selection.ts b/packages/tooling/vis/src/commands/run/run-selection.ts index 13ae6eea..dee2f234 100644 --- a/packages/tooling/vis/src/commands/run/run-selection.ts +++ b/packages/tooling/vis/src/commands/run/run-selection.ts @@ -1,4 +1,5 @@ import type { Toolbox } from "@visulima/cerebro"; +import { join } from "@visulima/path"; import { enforceProjectConstraints } from "@visulima/task-runner"; import { VisUserError } from "../../errors/vis-user-error"; @@ -247,6 +248,7 @@ export const narrowProjectSelection = async ({ * affected, so the run exits without executing anything. */ export const narrowToAffectedProjects = async ({ + cacheDirectory, fileOwners, logger, options, @@ -256,6 +258,7 @@ export const narrowToAffectedProjects = async ({ workspace, workspaceRoot, }: { + cacheDirectory?: string; fileOwners: DiscoveredWorkspace["fileOwners"]; logger: RunLogger; options: RunOptions; @@ -264,8 +267,18 @@ export const narrowToAffectedProjects = async ({ visConfig: Toolbox["visConfig"]; workspace: DiscoveredWorkspace["workspace"]; workspaceRoot: string; -}): Promise<{ affectedChangedFiles: string[] | undefined; projectNames: string[] } | undefined> => { +}): Promise< + | { + affectedChangedFiles: string[] | undefined; + externalInputHashes?: Record; + externalInputStatePath?: string; + projectNames: string[]; + } + | undefined +> => { let projectNames = projectNamesIn; + let externalInputHashes: Record | undefined; + let externalInputStatePath: string | undefined; // `--affected` narrows the selection in-process rather than re-invoking // `vis affected`. The old delegation round-trip could only forward the // handful of flags it enumerated, so --dry-run, --fail-fast, --partition @@ -282,15 +295,23 @@ export const narrowToAffectedProjects = async ({ uncommitted: options.uncommitted, upstream: options.upstream, }, - { fileOwners, projectGraph, projects: workspace.projects, workspaceRoot }, - { affectedIgnore: visConfig?.affectedIgnore, defaultBase: visConfig?.defaultBase }, + { fileOwners, namedInputs: visConfig?.namedInputs, projectGraph, projects: workspace.projects, workspaceRoot }, + { + affectedIgnore: visConfig?.affectedIgnore, + defaultBase: visConfig?.defaultBase, + externalInputStatePath: cacheDirectory ? join(cacheDirectory, "external-inputs.json") : undefined, + persistExternalInputState: false, + }, ); for (const note of affected.notes) { logger.info(`▸ ${note}`); } - if (affected.changedFiles.length === 0) { + externalInputHashes = affected.externalInputHashes; + externalInputStatePath = affected.externalInputStatePath; + + if (affected.changedFiles.length === 0 && (affected.changedExternalInputs?.length ?? 0) === 0) { logger.info("No files changed. Nothing to run."); return undefined; @@ -314,7 +335,7 @@ export const narrowToAffectedProjects = async ({ logger.info(`Affected projects: ${projectNames.join(", ")}`); } - return { affectedChangedFiles, projectNames }; + return { affectedChangedFiles, externalInputHashes, externalInputStatePath, projectNames }; }; /** @@ -325,6 +346,7 @@ export const narrowToAffectedProjects = async ({ */ export const resolveRunSelection = async ({ argument, + cacheDirectory, config, fileOwners, logger, @@ -337,6 +359,7 @@ export const resolveRunSelection = async ({ workspaceRoot, }: { argument: string[]; + cacheDirectory?: string; config: DiscoveredWorkspace["config"]; fileOwners: DiscoveredWorkspace["fileOwners"]; logger: RunLogger; @@ -349,6 +372,7 @@ export const resolveRunSelection = async ({ workspaceRoot: string; }): Promise< | { + affectedExternalInputState?: { hashes: Record; path: string }; affectedFiles: string[] | undefined; currentOs: ReturnType; forwardedArgs: string[]; @@ -392,6 +416,7 @@ export const resolveRunSelection = async ({ const forwardedArgs: string[] = argument.slice(1).map(String); const narrowed = await narrowToAffectedProjects({ + cacheDirectory, fileOwners, logger, options, @@ -406,7 +431,7 @@ export const resolveRunSelection = async ({ return undefined; } - const { affectedChangedFiles, projectNames } = narrowed; + const { affectedChangedFiles, externalInputHashes, externalInputStatePath, projectNames } = narrowed; const currentOs = detectCurrentOs(); @@ -423,6 +448,8 @@ export const resolveRunSelection = async ({ } const { projectsWithTarget, projectTargetIndex } = targetSelection; + const affectedExternalInputState + = externalInputStatePath && externalInputHashes ? { hashes: externalInputHashes, path: externalInputStatePath } : undefined; - return { affectedFiles, currentOs, forwardedArgs, projectsWithTarget, projectTargetIndex, target }; + return { affectedExternalInputState, affectedFiles, currentOs, forwardedArgs, projectsWithTarget, projectTargetIndex, target }; }; diff --git a/packages/tooling/vis/src/task/affected-selection.ts b/packages/tooling/vis/src/task/affected-selection.ts index f91bfc7a..20f0294c 100644 --- a/packages/tooling/vis/src/task/affected-selection.ts +++ b/packages/tooling/vis/src/task/affected-selection.ts @@ -1,11 +1,22 @@ import { spawnSync } from "node:child_process"; - -import type { AffectedResult, AffectedScope, ProjectConfiguration, ProjectGraph } from "@visulima/task-runner"; -import { getAffectedProjects } from "@visulima/task-runner"; +import { mkdir, readFile, rename, rm, writeFile } from "node:fs/promises"; + +import { dirname, join } from "@visulima/path"; +import type { + AffectedResult, + AffectedScope, + ExternalInputDefinition, + NamedInputs, + ProjectConfiguration, + ProjectGraph, + TargetConfiguration, +} from "@visulima/task-runner"; +import { collectExternalInputs, getAffectedProjects } from "@visulima/task-runner"; import isInCi from "is-in-ci"; import { VisUserError } from "../errors/vis-user-error"; import { resolveAffectedShas } from "../runtime/affected-shas"; +import { DEFAULT_WORKSPACE_CACHE_DIRECTORY } from "../util/vis-paths"; const VALID_SCOPES = new Set(["deep", "direct", "none"]); @@ -40,6 +51,23 @@ export interface SelectAffectedInput { /** Default base branch from `vis.config.ts#defaultBase`. */ defaultBase?: string; + /** + * Where the previous run's external-input hashes are stored. Defaults to + * `{workspaceRoot}/node_modules/.cache/vis/external-inputs.json` so the + * snapshot lives with derived cache state and is cleaned with it. + * Callers that resolved a custom cache directory should pass it here. + */ + externalInputStatePath?: string; + + /** + * Whether to write the external-input snapshot immediately. Defaults to + * `true` for selection-only commands (`vis affected --sparse-checkout`). + * `vis run --affected` sets this to `false` and persists after a + * successful run, so a failed run cannot advance the baseline past input + * state the tasks never processed. + */ + persistExternalInputState?: boolean; + /** Injectable for tests; defaults to a real `git` invocation. */ readWorkingTreeChanges?: (workspaceRoot: string) => string[]; @@ -48,6 +76,13 @@ export interface SelectAffectedInput { } export interface SelectAffectedResult extends AffectedResult { + /** + * Location of the external-input snapshot. Present when the selection + * declared external inputs; `vis run` uses it to persist after a + * successful run. + */ + externalInputStatePath?: string; + /** Human-readable provenance lines describing how base/head were picked. */ notes: string[]; @@ -116,6 +151,42 @@ const defaultReadWorkingTreeChanges = (workspaceRoot: string): string[] => { return parsePorcelainStatus(result.stdout); }; +const readExternalInputHashes = async (statePath: string): Promise | undefined> => { + try { + const content = await readFile(statePath, "utf8"); + const parsed: unknown = JSON.parse(content); + + return parsed !== null && typeof parsed === "object" && !Array.isArray(parsed) ? (parsed as Record) : undefined; + } catch { + // No snapshot yet (or an unreadable/corrupt one): affected detection + // treats every external input as changed, which is the safe answer. + return undefined; + } +}; + +export const persistExternalInputHashes = async (statePath: string, hashes: Record): Promise => { + const temporaryPath = `${statePath}.${process.pid}.tmp`; + + try { + await mkdir(dirname(statePath), { recursive: true }); + await writeFile(temporaryPath, JSON.stringify(hashes)); + await rename(temporaryPath, statePath); + } catch { + // Best-effort: a read-only workspace must not fail affected detection. + await rm(temporaryPath, { force: true }).catch(() => {}); + } +}; + +/** + * Drops the external-input snapshot after a failure that happened outside the + * regular task-results map (for example a persistent task exiting non-zero). + * The next selection then treats every external input as changed, which is the + * safe outcome. + */ +export const invalidateExternalInputState = async (statePath: string): Promise => { + await rm(statePath, { force: true }).catch(() => {}); +}; + /** * Resolve which projects are affected, applying the same base/head * resolution, scope validation and working-tree handling regardless of @@ -125,7 +196,14 @@ const defaultReadWorkingTreeChanges = (workspaceRoot: string): string[] => { */ export const selectAffectedProjects = async ( options: AffectedSelectionOptions, - workspace: { fileOwners?: Record; projectGraph: ProjectGraph; projects: Record; workspaceRoot: string }, + workspace: { + fileOwners?: Record; + namedInputs?: NamedInputs; + projectGraph: ProjectGraph; + projects: Record; + targetDefaults?: Record>; + workspaceRoot: string; + }, input: SelectAffectedInput = {}, ): Promise => { const downstreamValue = options.downstream ?? "deep"; @@ -139,7 +217,7 @@ export const selectAffectedProjects = async ( throw new VisUserError(`Invalid --upstream value: "${upstreamValue}". Must be "none", "direct", or "deep".`); } - const { fileOwners, projectGraph, projects, workspaceRoot } = workspace; + const { fileOwners, namedInputs, projectGraph, projects, targetDefaults, workspaceRoot } = workspace; const notes: string[] = []; let { base } = options; @@ -209,22 +287,41 @@ export const selectAffectedProjects = async ( } } + const externalInputs: ExternalInputDefinition[] = collectExternalInputs(workspaceRoot, projects, { namedInputs, targetDefaults }); + const externalInputStatePath = input.externalInputStatePath ?? join(workspaceRoot, DEFAULT_WORKSPACE_CACHE_DIRECTORY, "external-inputs.json"); + const previousExternalInputHashes = externalInputs.length > 0 ? await readExternalInputHashes(externalInputStatePath) : undefined; + const result = await getAffectedProjects({ additionalChangedFiles, base, downstream: downstreamValue as AffectedScope, + externalInputs, fileOwners, head, ignoredFiles: input.affectedIgnore, + previousExternalInputHashes, projectGraph, projects, upstream: upstreamValue as AffectedScope, workspaceRoot, }); + if (externalInputs.length > 0 && result.externalInputHashes && input.persistExternalInputState !== false) { + await persistExternalInputHashes(externalInputStatePath, result.externalInputHashes); + } + + if (result.changedExternalInputs && result.changedExternalInputs.length > 0) { + notes.push(`including ${result.changedExternalInputs.length} changed external input(s) outside the workspace`); + } + if (result.ignoredFiles.length > 0) { notes.push(`ignoring ${result.ignoredFiles.length} changed path(s) outside any project matching affectedIgnore`); } - return { ...result, notes, uncommittedFileCount: additionalChangedFiles.length }; + return { + ...result, + externalInputStatePath: externalInputs.length > 0 ? externalInputStatePath : undefined, + notes, + uncommittedFileCount: additionalChangedFiles.length, + }; };