fix: bound workspace Git scans (#11572)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Workspaces let users and agents inspect files that belong to an issue > - Changed-file views use full-tree Git status scans > - Many issue views could start those scans at the same time and make the server unresponsive > - Route-level limits did not protect the process or coalesce work for one repository > - This pull request adds one bounded scheduler for every expensive workspace Git scan > - It also starts browser scans only when the file panel is open and visible > - The benefit is bounded child-process use and responsive health checks during request storms ## Linked Issues or Issue Description **What happened?** Many changed-file requests could start full `git status --porcelain=v1 -z --untracked-files=all` scans at the same time. One production incident produced about 270 direct Git child processes. The Node process stayed alive but stopped answering health requests in time. **Expected behavior** Paperclip must bound expensive Git work across all companies, actors, issues, repositories, and browser tabs. Duplicate requests for one worktree must share work. Excess requests must fail fast with a retryable response. Hidden or closed file panels must not start scans. **Steps to reproduce** 1. Open changed-file views for many issue and actor keys. 2. Send requests for two large workspace roots at the same time. 3. Observe that route-level limiter keys allow many full Git scans to run together. 4. Observe delayed health responses and accumulated Git children. **Paperclip version or commit** Reproduced on master before commit `43ab441f0f`. **Deployment mode** Self-hosted server with local workspace repositories. ## What Changed - Add a process-wide scheduler with configurable concurrency, queue capacity, timeout, and cache TTL. - Add fair admission, a bounded queue, canonical worktree keys, single-flight joins, and bounded result caching. - Add subprocess timeouts, TERM-to-KILL escalation, bounded output, waiter cancellation, and slot cleanup. - Route full-tree status work from file resources, workspace runtime, execution workspaces, and adapter overlay sync through the scheduler. - Return stable retryable `503` and `504` error codes for saturation and timeout. - Add structured logs with safe workspace hashes, durations, queue state, cache use, joins, and terminal outcomes. - Gate UI queries on panel and document visibility. Cancel queries on close, hide, unmount, and workspace change. - Disable focus and reconnect bursts. Keep one explicit refresh action and a retryable unavailable state. - Document the 10-second default freshness tradeoff and all configuration variables. - Add unit, route, UI, adapter, and deterministic 500-request load coverage. ## Verification - `pnpm -r typecheck` - `pnpm build` - `pnpm check:token-gates` - `pnpm --filter @paperclipai/server exec vitest run src/services/workspace-git-operation-scheduler.test.ts src/__tests__/file-resources-git-scan-load.test.ts --reporter=dot` — 16 tests passed. - `pnpm --filter @paperclipai/ui exec vitest run src/components/WorkspaceFileBrowser.test.tsx src/lib/page-visibility.test.ts --reporter=dot` — 38 tests passed. - `pnpm --filter @paperclipai/adapter-utils exec vitest run src/git-workspace-sync.test.ts --reporter=dot` — 16 tests passed. - Existing file-resource, workspace-runtime, and execution-workspace regression selections passed. - Two cleanup safety regressions prove failed scans preserve the worktree before archive and at the final deletion fence. - Before: the incident produced about 270 Git children and health requests timed out. - After: 500 concurrent requests across 500 issue keys, 73 actors, and two roots started two underlying scans. Peak scan concurrency was 2. All 500 requests succeeded. Health p99 was 4.94 ms. The harness found zero unreaped children. - The full local Vitest run passed 4,267 tests. Ten existing fixed-port HTTPS exposure tests could not run because this host already owns Tailnet listeners on ports 42000 and 52000. Clean GitHub CI is the final full-suite result. - Latest-head GitHub CI passed all required test, typecheck, build, canary, e2e, policy, and security gates. - Greptile completed at 5/5 with zero unresolved comments, recommendations, or follow-ups. ## Risks - Changed-file results can be up to 10 seconds old by default. Explicit refresh remains available. - A full queue returns a retryable `503` instead of waiting without a bound. - A scan that exceeds the default 8-second deadline returns a retryable `504` and terminates its process group. - Operators can tune all limits with documented environment variables. Safe defaults protect local and shared servers. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex, GPT-5 family. The runtime does not expose the exact deployment ID or context-window size. High reasoning, tool use, and code execution were enabled. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
48f4ae16ac
commit
8087661bb8
|
|
@ -4,5 +4,11 @@ SERVE_UI=false
|
|||
BETTER_AUTH_SECRET=paperclip-dev-secret
|
||||
PAPERCLIP_TOOL_ACTION_SIGNING_SECRET=paperclip-dev-tool-action-signing-secret-change-me
|
||||
|
||||
# Process-wide protection for expensive full-tree workspace Git scans.
|
||||
# PAPERCLIP_WORKSPACE_GIT_SCAN_CONCURRENCY=2
|
||||
# PAPERCLIP_WORKSPACE_GIT_SCAN_QUEUE_CAPACITY=32
|
||||
# PAPERCLIP_WORKSPACE_GIT_SCAN_TIMEOUT_MS=8000
|
||||
# PAPERCLIP_WORKSPACE_GIT_SCAN_CACHE_TTL_MS=10000
|
||||
|
||||
# Discord webhook for daily merge digest (scripts/discord-daily-digest.sh)
|
||||
# DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/...
|
||||
|
|
|
|||
|
|
@ -425,6 +425,21 @@ Agent, project, environment, secret, skill, and workspace config edits are sampl
|
|||
|
||||
When effective run config changes, Paperclip may intentionally skip a saved adapter session, refresh persisted workspace runtime config, replace a reused execution workspace, or avoid reusing a sandbox/environment lease. Fresh execution can lose adapter-specific session, workspace, or sandbox state; correctness of the next run's config takes priority over continuity. Plain environment values affect freshness through value hashes; run result JSON and workspace operation logs expose only the non-sensitive freshness decision categories, without storing secret values, full env maps, provider credentials, or private path details.
|
||||
|
||||
## Workspace Git Scan Protection
|
||||
|
||||
Paperclip applies one process-wide scheduler to expensive host-side workspace Git enumeration, including changed-file browsing, runtime/finalization cleanliness guards, and adapter sandbox-sync snapshots. The scheduler defaults to two active scans and a bounded queue of 32. Identical scans of the same canonical worktree share one subprocess, while successful changed-file listings are cached for 10 seconds. Correctness-sensitive runtime guards bypass the result cache.
|
||||
|
||||
The cache intentionally trades up to a few seconds of changed-file freshness for stable server latency. The file browser retains an explicit refresh action, does not start its query while the panel or browser tab is hidden, and presents overloads as retryable failures rather than an empty workspace. A full queue returns `503` with code `workspace_git_scan_saturated`; a scan exceeding its wall-clock limit returns `504` with code `workspace_git_scan_timeout`. Both responses include `Retry-After: 1`.
|
||||
|
||||
Environment overrides:
|
||||
|
||||
- `PAPERCLIP_WORKSPACE_GIT_SCAN_CONCURRENCY` (default `2`, range `1`–`16`)
|
||||
- `PAPERCLIP_WORKSPACE_GIT_SCAN_QUEUE_CAPACITY` (default `32`, range `0`–`1024`)
|
||||
- `PAPERCLIP_WORKSPACE_GIT_SCAN_TIMEOUT_MS` (default `8000`, range `100`–`120000`)
|
||||
- `PAPERCLIP_WORKSPACE_GIT_SCAN_CACHE_TTL_MS` (default `10000`, range `0`–`60000`)
|
||||
|
||||
Structured `workspace_git_scan` logs expose the operation name, a non-reversible workspace-path hash, queue and execution durations, active/queued counts, cache and single-flight use, and terminal outcome. Saturation and timeout warnings are rate-limited so an overload does not create a second logging storm.
|
||||
|
||||
## Worktree-local Instances
|
||||
|
||||
When developing from multiple git worktrees, do not point two Paperclip servers at the same embedded PostgreSQL data directory.
|
||||
|
|
|
|||
|
|
@ -1,5 +1,11 @@
|
|||
# @paperclipai/adapter-utils
|
||||
|
||||
## Unreleased
|
||||
|
||||
### Patch Changes
|
||||
|
||||
- Allow the Paperclip host to route adapter sandbox-sync full-tree Git enumeration through its process-wide bounded scheduler.
|
||||
|
||||
## 0.3.1
|
||||
|
||||
### Patch Changes
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ import {
|
|||
readGitWorkspaceSnapshot,
|
||||
runLocalGit,
|
||||
sanitizeGitRemoteUrl,
|
||||
setExpensiveWorkspaceGitExecutor,
|
||||
withShallowGitWorkspaceClone,
|
||||
} from "./git-workspace-sync.js";
|
||||
|
||||
|
|
@ -28,6 +29,7 @@ describe("git workspace sync", () => {
|
|||
const cleanupDirs: string[] = [];
|
||||
|
||||
afterEach(async () => {
|
||||
setExpensiveWorkspaceGitExecutor(null);
|
||||
while (cleanupDirs.length > 0) {
|
||||
const dir = cleanupDirs.pop();
|
||||
if (!dir) continue;
|
||||
|
|
@ -35,6 +37,31 @@ describe("git workspace sync", () => {
|
|||
}
|
||||
});
|
||||
|
||||
it("delegates every host-side full-tree enumeration to the registered scheduler", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-git-scheduler-hook-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const repo = await createRepo(rootDir);
|
||||
await writeFile(path.join(repo, "untracked.txt"), "untracked\n", "utf8");
|
||||
const operations: string[] = [];
|
||||
setExpensiveWorkspaceGitExecutor(async (input) => {
|
||||
operations.push(input.operation);
|
||||
return await runLocalGit(input.localDir, [...input.args], {
|
||||
timeout: input.timeout,
|
||||
maxBuffer: input.maxBuffer,
|
||||
});
|
||||
});
|
||||
|
||||
const snapshot = await readGitWorkspaceSnapshot(repo);
|
||||
|
||||
expect(snapshot?.overlayPaths).toContain("untracked.txt");
|
||||
expect(operations.sort()).toEqual([
|
||||
"adapter_sync.deleted_files",
|
||||
"adapter_sync.ignored_files",
|
||||
"adapter_sync.overlay_diff",
|
||||
"adapter_sync.untracked_files",
|
||||
]);
|
||||
});
|
||||
|
||||
async function createRepo(rootDir: string): Promise<string> {
|
||||
const repo = path.join(rootDir, "repo");
|
||||
await mkdir(repo, { recursive: true });
|
||||
|
|
|
|||
|
|
@ -17,6 +17,29 @@ export interface GitWorkspaceSnapshot {
|
|||
ignoredPaths: string[];
|
||||
}
|
||||
|
||||
export interface ExpensiveWorkspaceGitInput {
|
||||
localDir: string;
|
||||
args: readonly string[];
|
||||
operation: string;
|
||||
timeout: number;
|
||||
maxBuffer: number;
|
||||
}
|
||||
|
||||
export type ExpensiveWorkspaceGitExecutor = (
|
||||
input: ExpensiveWorkspaceGitInput,
|
||||
) => Promise<GitCommandResult>;
|
||||
|
||||
let expensiveWorkspaceGitExecutor: ExpensiveWorkspaceGitExecutor | null = null;
|
||||
|
||||
/**
|
||||
* Lets a host process apply its process-wide admission policy to the adapter
|
||||
* package's full-tree Git walks. Standalone adapter-utils consumers retain the
|
||||
* existing timeout/buffer-bounded fallback.
|
||||
*/
|
||||
export function setExpensiveWorkspaceGitExecutor(executor: ExpensiveWorkspaceGitExecutor | null): void {
|
||||
expensiveWorkspaceGitExecutor = executor;
|
||||
}
|
||||
|
||||
export const GIT_ARCHIVE_EXCLUDES = [".git", ".git/*"] as const;
|
||||
|
||||
function shellQuote(value: string) {
|
||||
|
|
@ -53,6 +76,24 @@ export async function runLocalGit(
|
|||
});
|
||||
}
|
||||
|
||||
async function runExpensiveWorkspaceGit(
|
||||
localDir: string,
|
||||
args: string[],
|
||||
operation: string,
|
||||
options: { timeout: number; maxBuffer: number },
|
||||
): Promise<GitCommandResult> {
|
||||
if (expensiveWorkspaceGitExecutor) {
|
||||
return await expensiveWorkspaceGitExecutor({
|
||||
localDir,
|
||||
args,
|
||||
operation,
|
||||
timeout: options.timeout,
|
||||
maxBuffer: options.maxBuffer,
|
||||
});
|
||||
}
|
||||
return await runLocalGit(localDir, args, options);
|
||||
}
|
||||
|
||||
export async function readGitWorkspaceSnapshot(localDir: string): Promise<GitWorkspaceSnapshot | null> {
|
||||
try {
|
||||
const insideWorkTree = await runLocalGit(localDir, ["rev-parse", "--is-inside-work-tree"], {
|
||||
|
|
@ -72,19 +113,19 @@ export async function readGitWorkspaceSnapshot(localDir: string): Promise<GitWor
|
|||
timeout: 10_000,
|
||||
maxBuffer: 16 * 1024,
|
||||
}),
|
||||
runLocalGit(localDir, ["diff", "--name-only", "-z", "--diff-filter=ACMRTUXB", "HEAD", "--"], {
|
||||
runExpensiveWorkspaceGit(localDir, ["diff", "--name-only", "-z", "--diff-filter=ACMRTUXB", "HEAD", "--"], "adapter_sync.overlay_diff", {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 1024 * 1024,
|
||||
}),
|
||||
runLocalGit(localDir, ["ls-files", "--others", "--exclude-standard", "-z"], {
|
||||
runExpensiveWorkspaceGit(localDir, ["ls-files", "--others", "--exclude-standard", "-z"], "adapter_sync.untracked_files", {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 1024 * 1024,
|
||||
}),
|
||||
runLocalGit(localDir, ["diff", "--name-only", "-z", "--diff-filter=D", "HEAD", "--"], {
|
||||
runExpensiveWorkspaceGit(localDir, ["diff", "--name-only", "-z", "--diff-filter=D", "HEAD", "--"], "adapter_sync.deleted_files", {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 256 * 1024,
|
||||
}),
|
||||
runLocalGit(localDir, ["status", "--ignored", "--porcelain=v1", "-z", "--untracked-files=normal"], {
|
||||
runExpensiveWorkspaceGit(localDir, ["status", "--ignored", "--porcelain=v1", "-z", "--untracked-files=normal"], "adapter_sync.ignored_files", {
|
||||
timeout: 10_000,
|
||||
maxBuffer: 1024 * 1024,
|
||||
}),
|
||||
|
|
|
|||
|
|
@ -1,5 +1,11 @@
|
|||
# @paperclipai/server
|
||||
|
||||
## Unreleased
|
||||
|
||||
### Patch Changes
|
||||
|
||||
- Bound full-tree workspace Git scans with process-wide concurrency, queue, timeout, cancellation, coalescing, and short-lived changed-file caching. Saturated or timed-out changed-file requests now return a retryable degraded response, and hidden file-browser panels no longer initiate scans.
|
||||
|
||||
## 0.3.1
|
||||
|
||||
### Patch Changes
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ import {
|
|||
startRuntimeServicesForWorkspaceControl,
|
||||
stopRuntimeServicesForExecutionWorkspace,
|
||||
} from "../services/workspace-runtime.ts";
|
||||
import { workspaceGitOperationScheduler } from "../services/workspace-git-operation-scheduler.ts";
|
||||
|
||||
const execFileAsync = promisify(execFile);
|
||||
|
||||
|
|
@ -549,6 +550,54 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => {
|
|||
expect(workspace).toMatchObject({ status: "archived", cleanupReason: "issue_terminal" });
|
||||
}, 20_000);
|
||||
|
||||
it("fails closed before archive when git status inspection is unavailable", async () => {
|
||||
const seeded = await seedAncestryTerminalWorkspace();
|
||||
const statusSpy = vi.spyOn(workspaceGitOperationScheduler, "run")
|
||||
.mockRejectedValue(new Error("scan queue unavailable"));
|
||||
|
||||
try {
|
||||
const readiness = await svc.getCloseReadiness(seeded.executionWorkspaceId);
|
||||
expect(readiness).toMatchObject({
|
||||
state: "blocked",
|
||||
isDestructiveCloseAllowed: false,
|
||||
blockingReasons: [
|
||||
"Paperclip could not verify the workspace git status. Retry before destructive cleanup.",
|
||||
],
|
||||
});
|
||||
|
||||
const sweep = await svc.sweepTerminalWorkspaces();
|
||||
expect(sweep).toMatchObject({ archived: 0, skippedUndelivered: 1 });
|
||||
const [workspace] = await db
|
||||
.select({ status: executionWorkspaces.status })
|
||||
.from(executionWorkspaces)
|
||||
.where(eq(executionWorkspaces.id, seeded.executionWorkspaceId));
|
||||
expect(workspace?.status).toBe("active");
|
||||
await expect(fs.access(seeded.worktreePath)).resolves.toBeUndefined();
|
||||
} finally {
|
||||
statusSpy.mockRestore();
|
||||
}
|
||||
}, 20_000);
|
||||
|
||||
it("fails the final cleanup fence when a later git status scan is unavailable", async () => {
|
||||
const seeded = await seedAncestryTerminalWorkspace();
|
||||
const originalRun = workspaceGitOperationScheduler.run.bind(workspaceGitOperationScheduler);
|
||||
let statusScanCount = 0;
|
||||
const statusSpy = vi.spyOn(workspaceGitOperationScheduler, "run")
|
||||
.mockImplementation(async (input) => {
|
||||
statusScanCount += 1;
|
||||
if (statusScanCount > 1) throw new Error("scan timed out");
|
||||
return originalRun(input);
|
||||
});
|
||||
|
||||
try {
|
||||
const sweep = await svc.sweepTerminalWorkspaces();
|
||||
expect(sweep).toMatchObject({ archived: 0, cleanupFailed: 1 });
|
||||
await expect(fs.access(seeded.worktreePath)).resolves.toBeUndefined();
|
||||
} finally {
|
||||
statusSpy.mockRestore();
|
||||
}
|
||||
}, 20_000);
|
||||
|
||||
it("skips a sweep that starts while another sweep runs", async () => {
|
||||
// The scheduler can start a second sweep before the first one finishes. The
|
||||
// sweeps share the cursor and the boundary. A concurrent sweep must skip
|
||||
|
|
|
|||
|
|
@ -0,0 +1,252 @@
|
|||
import express from "express";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import request from "supertest";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { errorHandler } from "../middleware/index.js";
|
||||
import {
|
||||
createFileResourceListLimiter,
|
||||
fileResourceRoutes,
|
||||
type WorkspaceFileResourceService,
|
||||
} from "../routes/file-resources.js";
|
||||
import {
|
||||
createWorkspaceGitOperationScheduler,
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES,
|
||||
WorkspaceGitScanError,
|
||||
type WorkspaceGitRunner,
|
||||
} from "../services/workspace-git-operation-scheduler.js";
|
||||
vi.mock("../services/activity-log.js", () => ({
|
||||
logActivity: vi.fn(async () => undefined),
|
||||
}));
|
||||
|
||||
function unavailableMethods(): Pick<WorkspaceFileResourceService, "availability" | "resolve" | "readContent" | "prepareDownload"> {
|
||||
return {
|
||||
availability: vi.fn(async () => { throw new Error("not used"); }),
|
||||
resolve: vi.fn(async () => { throw new Error("not used"); }),
|
||||
readContent: vi.fn(async () => { throw new Error("not used"); }),
|
||||
prepareDownload: vi.fn(async () => { throw new Error("not used"); }),
|
||||
};
|
||||
}
|
||||
|
||||
function availableList(input: { mode?: "all" | "recent" | "changed" | null } = {}) {
|
||||
return {
|
||||
kind: "workspace_file_list" as const,
|
||||
state: "available" as const,
|
||||
workspace: {
|
||||
provider: "local_fs" as const,
|
||||
workspaceLabel: "Workspace",
|
||||
workspaceKind: "project_workspace" as const,
|
||||
workspaceId: "11111111-1111-4111-8111-111111111111",
|
||||
},
|
||||
query: {
|
||||
workspace: "auto" as const,
|
||||
mode: input.mode ?? "changed",
|
||||
q: null,
|
||||
limit: 25,
|
||||
},
|
||||
items: [],
|
||||
scannedCount: 0,
|
||||
truncated: false,
|
||||
};
|
||||
}
|
||||
|
||||
function createLoadApp(db: Db, companyId: string, service: WorkspaceFileResourceService) {
|
||||
const app = express();
|
||||
app.get("/api/health", (_req, res) => res.json({ status: "ok" }));
|
||||
app.use((req, _res, next) => {
|
||||
req.actor = {
|
||||
type: "board",
|
||||
userId: String(req.headers["x-test-actor"] ?? "board-user"),
|
||||
companyIds: [companyId],
|
||||
source: "session",
|
||||
isInstanceAdmin: false,
|
||||
};
|
||||
next();
|
||||
});
|
||||
app.use("/api", fileResourceRoutes(db, {
|
||||
service,
|
||||
listLimiter: createFileResourceListLimiter({
|
||||
maxConcurrent: 2,
|
||||
maxRequests: 10,
|
||||
windowMs: 60_000,
|
||||
}),
|
||||
}));
|
||||
app.use(errorHandler);
|
||||
return app;
|
||||
}
|
||||
|
||||
describe("workspace Git scan route load regression", () => {
|
||||
const db = {} as Db;
|
||||
const companyId = "11111111-1111-4111-8111-111111111111";
|
||||
let tempRoot: string;
|
||||
|
||||
beforeAll(async () => {
|
||||
tempRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-file-scan-load-"));
|
||||
await Promise.all([
|
||||
fs.mkdir(path.join(tempRoot, "repository-a")),
|
||||
fs.mkdir(path.join(tempRoot, "repository-b")),
|
||||
]);
|
||||
}, 60_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await fs.rm(tempRoot, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
it("keeps 500 cross-actor/cross-issue requests globally bounded while health stays responsive", async () => {
|
||||
let releaseScans!: () => void;
|
||||
const scanGate = new Promise<void>((resolve) => {
|
||||
releaseScans = resolve;
|
||||
});
|
||||
let active = 0;
|
||||
let peakActive = 0;
|
||||
let runnerCalls = 0;
|
||||
const runner: WorkspaceGitRunner = async () => {
|
||||
runnerCalls += 1;
|
||||
active += 1;
|
||||
peakActive = Math.max(peakActive, active);
|
||||
await scanGate;
|
||||
active -= 1;
|
||||
return { stdout: "", stderr: "" };
|
||||
};
|
||||
const scheduler = createWorkspaceGitOperationScheduler({
|
||||
concurrency: 2,
|
||||
queueCapacity: 4,
|
||||
runner,
|
||||
defaultCacheTtlMs: 0,
|
||||
});
|
||||
const roots = [path.join(tempRoot, "repository-a"), path.join(tempRoot, "repository-b")];
|
||||
const service: WorkspaceFileResourceService = {
|
||||
getIssue: vi.fn(async () => ({ companyId })),
|
||||
list: vi.fn(async (issueId, input, opts) => {
|
||||
const numericId = Number(issueId.slice("issue-".length));
|
||||
await scheduler.run({
|
||||
workspacePath: roots[numericId % roots.length]!,
|
||||
args: ["status", "--porcelain=v1", "-z", "--untracked-files=all"],
|
||||
operation: "route_load.changed_files",
|
||||
fairnessKeys: opts?.scanContext?.fairnessKeys,
|
||||
signal: opts?.scanContext?.signal,
|
||||
cacheTtlMs: 0,
|
||||
});
|
||||
return availableList(input);
|
||||
}),
|
||||
...unavailableMethods(),
|
||||
};
|
||||
const app = createLoadApp(db, companyId, service);
|
||||
const pendingResponses = Array.from({ length: 500 }, (_, index) => request(app)
|
||||
.get(`/api/issues/issue-${index}/file-resources/list`)
|
||||
.set("x-test-actor", `actor-${index % 73}`)
|
||||
.query({ mode: "changed" })
|
||||
.then((response) => response));
|
||||
|
||||
await vi.waitFor(
|
||||
() => expect(scheduler.snapshot().totals.singleFlightJoins).toBe(498),
|
||||
{ timeout: 15_000, interval: 20 },
|
||||
);
|
||||
const loadedSnapshot = scheduler.snapshot();
|
||||
expect(loadedSnapshot).toMatchObject({ activeCount: 2, queuedCount: 0, inFlightCount: 2 });
|
||||
|
||||
const healthLatencies: number[] = [];
|
||||
for (let index = 0; index < 25; index += 1) {
|
||||
const startedAt = performance.now();
|
||||
const response = await request(app).get("/api/health");
|
||||
healthLatencies.push(performance.now() - startedAt);
|
||||
expect(response.status).toBe(200);
|
||||
}
|
||||
healthLatencies.sort((left, right) => left - right);
|
||||
const healthP99Ms = healthLatencies[Math.ceil(healthLatencies.length * 0.99) - 1]!;
|
||||
expect(healthP99Ms).toBeLessThan(250);
|
||||
|
||||
releaseScans();
|
||||
const responses = await Promise.all(pendingResponses);
|
||||
const outcomeCounts = responses.reduce<Record<number, number>>((counts, response) => {
|
||||
counts[response.status] = (counts[response.status] ?? 0) + 1;
|
||||
return counts;
|
||||
}, {});
|
||||
expect(outcomeCounts).toEqual({ 200: 500 });
|
||||
expect({ runnerCalls, peakActive }).toEqual({ runnerCalls: 2, peakActive: 2 });
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, queuedCount: 0, inFlightCount: 0 });
|
||||
|
||||
console.info("workspace Git scan regression metrics", {
|
||||
requests: responses.length,
|
||||
repositories: roots.length,
|
||||
underlyingScans: runnerCalls,
|
||||
peakActive,
|
||||
peakQueued: loadedSnapshot.queuedCount,
|
||||
outcomes: outcomeCounts,
|
||||
healthP99Ms: Math.round(healthP99Ms * 100) / 100,
|
||||
unreapedChildCount: 0,
|
||||
});
|
||||
}, 60_000);
|
||||
|
||||
it.each([
|
||||
[WORKSPACE_GIT_SCAN_ERROR_CODES.saturated, 503],
|
||||
[WORKSPACE_GIT_SCAN_ERROR_CODES.timeout, 504],
|
||||
] as const)("returns stable retryable %s responses", async (code, status) => {
|
||||
const service: WorkspaceFileResourceService = {
|
||||
getIssue: vi.fn(async () => ({ companyId })),
|
||||
list: vi.fn(async () => {
|
||||
throw new WorkspaceGitScanError(code, "Changed files temporarily unavailable");
|
||||
}),
|
||||
...unavailableMethods(),
|
||||
};
|
||||
const app = createLoadApp(db, companyId, service);
|
||||
|
||||
const response = await request(app)
|
||||
.get(`/api/issues/${code}/file-resources/list`)
|
||||
.query({ mode: "changed" });
|
||||
|
||||
expect(response.status).toBe(status);
|
||||
expect(response.headers["retry-after"]).toBe("1");
|
||||
expect(response.body).toMatchObject({
|
||||
code,
|
||||
details: { code, retryable: true },
|
||||
});
|
||||
});
|
||||
|
||||
it("cancels the underlying scan and releases its slot when the HTTP client disconnects", async () => {
|
||||
let runnerAborted = false;
|
||||
const scheduler = createWorkspaceGitOperationScheduler({
|
||||
concurrency: 1,
|
||||
runner: ({ signal }) => new Promise((_resolve, reject) => {
|
||||
signal.addEventListener("abort", () => {
|
||||
runnerAborted = true;
|
||||
reject(new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled,
|
||||
"cancelled",
|
||||
));
|
||||
}, { once: true });
|
||||
}),
|
||||
});
|
||||
const service: WorkspaceFileResourceService = {
|
||||
getIssue: vi.fn(async () => ({ companyId })),
|
||||
list: vi.fn(async (_issueId, input, opts) => {
|
||||
await scheduler.run({
|
||||
workspacePath: path.join(tempRoot, "repository-a"),
|
||||
args: ["status", "--porcelain=v1", "-z", "--untracked-files=all"],
|
||||
operation: "route_disconnect.changed_files",
|
||||
signal: opts?.scanContext?.signal,
|
||||
cacheTtlMs: 0,
|
||||
});
|
||||
return availableList(input);
|
||||
}),
|
||||
...unavailableMethods(),
|
||||
};
|
||||
const app = createLoadApp(db, companyId, service);
|
||||
const pending = request(app)
|
||||
.get("/api/issues/disconnected/file-resources/list")
|
||||
.query({ mode: "changed" });
|
||||
const outcome = pending.then(
|
||||
(response) => response,
|
||||
(error: unknown) => error,
|
||||
);
|
||||
await vi.waitFor(() => expect(scheduler.snapshot().activeCount).toBe(1));
|
||||
|
||||
pending.abort();
|
||||
await outcome;
|
||||
|
||||
await vi.waitFor(() => expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, inFlightCount: 0 }));
|
||||
expect(runnerAborted).toBe(true);
|
||||
});
|
||||
});
|
||||
|
|
@ -1,6 +1,6 @@
|
|||
import { createReadStream } from "node:fs";
|
||||
import { pipeline } from "node:stream/promises";
|
||||
import { Router } from "express";
|
||||
import { Router, type Request, type Response } from "express";
|
||||
import { ZodError } from "zod";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
|
|
@ -17,6 +17,10 @@ import { badRequest, HttpError, notFound, unprocessable } from "../errors.js";
|
|||
import { workspaceFileResourceService } from "../services/index.js";
|
||||
import { assertBoard, getActorInfo, hasCompanyAccess } from "./authz.js";
|
||||
import { logActivity } from "../services/activity-log.js";
|
||||
import {
|
||||
isWorkspaceGitScanError,
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES,
|
||||
} from "../services/workspace-git-operation-scheduler.js";
|
||||
|
||||
export type WorkspaceFileResourceService = {
|
||||
getIssue(issueId: string): Promise<{ companyId: string }>;
|
||||
|
|
@ -34,7 +38,10 @@ export type WorkspaceFileResourceService = {
|
|||
q?: string | null;
|
||||
limit?: number | null;
|
||||
offset?: number | null;
|
||||
}, opts?: { issue?: Awaited<ReturnType<WorkspaceFileResourceService["getIssue"]>> }): Promise<WorkspaceFileListResponse>;
|
||||
}, opts?: {
|
||||
issue?: Awaited<ReturnType<WorkspaceFileResourceService["getIssue"]>>;
|
||||
scanContext?: { signal?: AbortSignal; fairnessKeys?: readonly string[] };
|
||||
}): Promise<WorkspaceFileListResponse>;
|
||||
resolve(
|
||||
issueId: string,
|
||||
input: { path: string; workspace?: "auto" | "execution" | "project" | null; projectId?: string | null; workspaceId?: string | null },
|
||||
|
|
@ -133,6 +140,23 @@ function limiterKey(companyId: string, actorId: string, issueId: string) {
|
|||
return `${companyId}:${actorId}:${issueId}`;
|
||||
}
|
||||
|
||||
function requestAbortController(req: Request, res: Response) {
|
||||
const controller = new AbortController();
|
||||
const abort = () => controller.abort();
|
||||
const abortIfResponseIncomplete = () => {
|
||||
if (!res.writableEnded) abort();
|
||||
};
|
||||
req.once("aborted", abort);
|
||||
res.once("close", abortIfResponseIncomplete);
|
||||
return {
|
||||
signal: controller.signal,
|
||||
cleanup() {
|
||||
req.off("aborted", abort);
|
||||
res.off("close", abortIfResponseIncomplete);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function parseBooleanQuery(value: unknown) {
|
||||
return value === true || value === "true" || value === "1";
|
||||
}
|
||||
|
|
@ -569,8 +593,19 @@ export function fileResourceRoutes(db: Db, opts: {
|
|||
throw error;
|
||||
}
|
||||
|
||||
const requestAbort = requestAbortController(req, res);
|
||||
try {
|
||||
const result = await svc.list(req.params.issueId, query, { issue });
|
||||
const result = await svc.list(req.params.issueId, query, {
|
||||
issue,
|
||||
scanContext: {
|
||||
signal: requestAbort.signal,
|
||||
fairnessKeys: [
|
||||
`company:${issue.companyId}`,
|
||||
`actor:${actor.actorId}`,
|
||||
`issue:${req.params.issueId}`,
|
||||
],
|
||||
},
|
||||
});
|
||||
await logActivity(db, {
|
||||
companyId: issue.companyId,
|
||||
actorType: actor.actorType,
|
||||
|
|
@ -597,6 +632,17 @@ export function fileResourceRoutes(db: Db, opts: {
|
|||
});
|
||||
res.json(result);
|
||||
} catch (error) {
|
||||
if (
|
||||
isWorkspaceGitScanError(error) &&
|
||||
error.code === WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled &&
|
||||
requestAbort.signal.aborted &&
|
||||
(res.destroyed || res.writableEnded)
|
||||
) {
|
||||
return;
|
||||
}
|
||||
if (isWorkspaceGitScanError(error) && error.status >= 500) {
|
||||
res.setHeader("Retry-After", "1");
|
||||
}
|
||||
await logListDeniedAttempt({
|
||||
companyId: issue.companyId,
|
||||
actor,
|
||||
|
|
@ -607,6 +653,7 @@ export function fileResourceRoutes(db: Db, opts: {
|
|||
});
|
||||
throw error;
|
||||
} finally {
|
||||
requestAbort.cleanup();
|
||||
release?.();
|
||||
}
|
||||
});
|
||||
|
|
|
|||
|
|
@ -54,6 +54,7 @@ import {
|
|||
import { visibleIssueCondition } from "./issue-visibility.js";
|
||||
import { createGitRemoteAuthProvider } from "./git-credentials.js";
|
||||
import { readProjectWorkspaceRuntimeConfig } from "./project-workspace-runtime-config.js";
|
||||
import { workspaceGitOperationScheduler } from "./workspace-git-operation-scheduler.js";
|
||||
import {
|
||||
listCurrentRuntimeServicesForExecutionWorkspaces,
|
||||
listCurrentRuntimeServicesForProjectWorkspaces,
|
||||
|
|
@ -374,6 +375,21 @@ async function runGit(args: string[], cwd: string) {
|
|||
return await execFileAsync("git", ["-C", cwd, ...args], { cwd });
|
||||
}
|
||||
|
||||
async function runExpensiveGitStatus(input: {
|
||||
args: readonly string[];
|
||||
cwd: string;
|
||||
operation: string;
|
||||
fairnessKeys?: readonly string[];
|
||||
}) {
|
||||
return workspaceGitOperationScheduler.run({
|
||||
workspacePath: input.cwd,
|
||||
args: input.args,
|
||||
operation: input.operation,
|
||||
fairnessKeys: input.fairnessKeys,
|
||||
cacheTtlMs: 0,
|
||||
});
|
||||
}
|
||||
|
||||
async function readGitStdout(args: string[], cwd: string): Promise<string | null> {
|
||||
const output = await runGit(args, cwd);
|
||||
return output.stdout.trim() || null;
|
||||
|
|
@ -502,7 +518,15 @@ async function inspectExecutionWorkspaceBranchForReconcile(
|
|||
throw unprocessable("Execution workspace is detached; Paperclip cannot reconcile it to a branch name");
|
||||
}
|
||||
|
||||
const status = await runGit(["status", "--porcelain", "--untracked-files=all"], worktreePath)
|
||||
const status = await runExpensiveGitStatus({
|
||||
args: ["status", "--porcelain", "--untracked-files=all"],
|
||||
cwd: worktreePath,
|
||||
operation: "execution_workspaces.branch_reconcile_status",
|
||||
fairnessKeys: [
|
||||
`workspace:${workspace.id}`,
|
||||
...(workspace.sourceIssueId ? [`issue:${workspace.sourceIssueId}`] : []),
|
||||
],
|
||||
})
|
||||
.then((output) => output.stdout)
|
||||
.catch(() => null);
|
||||
const statusLines = status === null
|
||||
|
|
@ -739,6 +763,7 @@ async function quarantineRestoreDirtyWorkspaceBranch(input: {
|
|||
async function inspectGitCloseReadiness(workspace: ExecutionWorkspace): Promise<{
|
||||
git: ExecutionWorkspaceCloseGitReadiness | null;
|
||||
warnings: string[];
|
||||
statusInspectionSucceeded: boolean;
|
||||
}> {
|
||||
const warnings: string[] = [];
|
||||
const workspacePath = readNullableString(workspace.providerRef) ?? readNullableString(workspace.cwd);
|
||||
|
|
@ -748,12 +773,12 @@ async function inspectGitCloseReadiness(workspace: ExecutionWorkspace): Promise<
|
|||
Boolean(workspace.repoUrl || workspace.baseRef || workspace.branchName || workspacePath);
|
||||
|
||||
if (!expectsGitInspection) {
|
||||
return { git: null, warnings };
|
||||
return { git: null, warnings, statusInspectionSucceeded: true };
|
||||
}
|
||||
|
||||
if (!workspacePath) {
|
||||
warnings.push("Workspace has no local path, so Paperclip cannot inspect git status before close.");
|
||||
return { git: null, warnings };
|
||||
return { git: null, warnings, statusInspectionSucceeded: false };
|
||||
}
|
||||
|
||||
if (!(await pathExists(workspacePath))) {
|
||||
|
|
@ -774,6 +799,7 @@ async function inspectGitCloseReadiness(workspace: ExecutionWorkspace): Promise<
|
|||
createdByRuntime,
|
||||
},
|
||||
warnings,
|
||||
statusInspectionSucceeded: true,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -797,9 +823,19 @@ async function inspectGitCloseReadiness(workspace: ExecutionWorkspace): Promise<
|
|||
|
||||
let dirtyEntryCount = 0;
|
||||
let untrackedEntryCount = 0;
|
||||
let statusInspectionSucceeded = false;
|
||||
if (repoRoot) {
|
||||
try {
|
||||
const statusOutput = (await runGit(["status", "--porcelain=v1", "--untracked-files=all"], workspacePath)).stdout;
|
||||
const statusOutput = (await runExpensiveGitStatus({
|
||||
args: ["status", "--porcelain=v1", "--untracked-files=all"],
|
||||
cwd: workspacePath,
|
||||
operation: "execution_workspaces.close_readiness_status",
|
||||
fairnessKeys: [
|
||||
`company:${workspace.companyId}`,
|
||||
`workspace:${workspace.id}`,
|
||||
...(workspace.sourceIssueId ? [`issue:${workspace.sourceIssueId}`] : []),
|
||||
],
|
||||
})).stdout;
|
||||
for (const line of statusOutput.split(/\r?\n/)) {
|
||||
if (!line) continue;
|
||||
if (line.startsWith("??")) {
|
||||
|
|
@ -808,6 +844,7 @@ async function inspectGitCloseReadiness(workspace: ExecutionWorkspace): Promise<
|
|||
}
|
||||
dirtyEntryCount += 1;
|
||||
}
|
||||
statusInspectionSucceeded = true;
|
||||
} catch (error) {
|
||||
warnings.push(
|
||||
`Could not read git working tree status for "${workspacePath}": ${error instanceof Error ? error.message : String(error)}`,
|
||||
|
|
@ -862,6 +899,7 @@ async function inspectGitCloseReadiness(workspace: ExecutionWorkspace): Promise<
|
|||
createdByRuntime,
|
||||
},
|
||||
warnings,
|
||||
statusInspectionSucceeded,
|
||||
};
|
||||
}
|
||||
|
||||
|
|
@ -1368,6 +1406,9 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
|||
readGitStdout(["rev-parse", "HEAD"], workspacePath).catch(() => null),
|
||||
readGitStdout(["symbolic-ref", "--quiet", "--short", "HEAD"], workspacePath).catch(() => null),
|
||||
]);
|
||||
if (!current.statusInspectionSucceeded) {
|
||||
throw new Error("Refusing terminal workspace cleanup because the git status could not be verified");
|
||||
}
|
||||
if (
|
||||
!current.git?.repoRoot
|
||||
|| current.git.hasDirtyTrackedFiles
|
||||
|
|
@ -2206,10 +2247,17 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
|||
|
||||
const executionWorkspace = toExecutionWorkspace(workspace, runtimeServices);
|
||||
const config = readExecutionWorkspaceConfig((workspace.metadata as Record<string, unknown> | null) ?? null);
|
||||
const { git, warnings: gitWarnings } = await inspectGitCloseReadiness(executionWorkspace);
|
||||
const {
|
||||
git,
|
||||
warnings: gitWarnings,
|
||||
statusInspectionSucceeded,
|
||||
} = await inspectGitCloseReadiness(executionWorkspace);
|
||||
const { deliveryState } = await assessDelivery(workspace, git);
|
||||
const warnings = [...gitWarnings];
|
||||
const blockingReasons: string[] = [];
|
||||
if (!statusInspectionSucceeded) {
|
||||
blockingReasons.push("Paperclip could not verify the workspace git status. Retry before destructive cleanup.");
|
||||
}
|
||||
const isSharedWorkspace = executionWorkspace.mode === "shared_workspace";
|
||||
const workspacePath = readNullableString(executionWorkspace.providerRef) ?? readNullableString(executionWorkspace.cwd);
|
||||
const resolvedWorkspacePath = workspacePath ? path.resolve(workspacePath) : null;
|
||||
|
|
@ -2480,7 +2528,11 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
|
|||
|
||||
for (const workspace of candidates) {
|
||||
const executionWorkspace = toExecutionWorkspace(workspace);
|
||||
const { git } = await inspectGitCloseReadiness(executionWorkspace);
|
||||
const { git, statusInspectionSucceeded } = await inspectGitCloseReadiness(executionWorkspace);
|
||||
if (!statusInspectionSucceeded) {
|
||||
result.skippedUndelivered += 1;
|
||||
continue;
|
||||
}
|
||||
const assessment = await assessDelivery(workspace, git);
|
||||
const reopenPending = metadataHasReopenPendingConsumption(
|
||||
workspace.metadata as Record<string, unknown> | null,
|
||||
|
|
|
|||
|
|
@ -169,6 +169,12 @@ export {
|
|||
type WorkspaceRuntimeLeaseService,
|
||||
} from "./workspace-runtime-leases.js";
|
||||
export { workspaceFileResourceService } from "./workspace-file-resources.js";
|
||||
export {
|
||||
createWorkspaceGitOperationScheduler,
|
||||
getWorkspaceGitOperationSchedulerSnapshot,
|
||||
workspaceGitOperationScheduler,
|
||||
type WorkspaceGitSchedulerSnapshot,
|
||||
} from "./workspace-git-operation-scheduler.js";
|
||||
export { workProductService } from "./work-products.js";
|
||||
export {
|
||||
logActivity,
|
||||
|
|
|
|||
|
|
@ -1,7 +1,5 @@
|
|||
import { execFile } from "node:child_process";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { promisify } from "node:util";
|
||||
import { and, desc, eq, inArray, isNull } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { executionWorkspaces, issues, projects, projectWorkspaces } from "@paperclipai/db";
|
||||
|
|
@ -20,6 +18,11 @@ import type {
|
|||
WorkspaceFileWorkspaceKind,
|
||||
} from "@paperclipai/shared";
|
||||
import { HttpError, notFound, unprocessable } from "../errors.js";
|
||||
import {
|
||||
isWorkspaceGitScanError,
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES,
|
||||
workspaceGitOperationScheduler,
|
||||
} from "./workspace-git-operation-scheduler.js";
|
||||
|
||||
export const WORKSPACE_FILE_TEXT_MAX_BYTES = 512 * 1024;
|
||||
export const WORKSPACE_FILE_MEDIA_MAX_BYTES = 10 * 1024 * 1024;
|
||||
|
|
@ -31,9 +34,13 @@ const MAX_RELATIVE_PATH_BYTES = 4096;
|
|||
const TEXT_SNIFF_BYTES = 4096;
|
||||
const MAX_LIST_DEPTH = 20;
|
||||
const GIT_STATUS_MAX_BUFFER_BYTES = 1024 * 1024;
|
||||
const execFileAsync = promisify(execFile);
|
||||
const LOCAL_PROJECT_WORKSPACE_SOURCE_TYPES = new Set(["local_path", "non_git_path", "git_repo", "git_worktree"]);
|
||||
|
||||
export interface WorkspaceFileScanContext {
|
||||
signal?: AbortSignal;
|
||||
fairnessKeys?: readonly string[];
|
||||
}
|
||||
|
||||
const DENIED_SEGMENTS = new Set([
|
||||
".git",
|
||||
".paperclip",
|
||||
|
|
@ -976,16 +983,29 @@ async function listChangedWorkspaceFiles(input: {
|
|||
normalizedQuery: string | null;
|
||||
limit: number;
|
||||
offset: number;
|
||||
scanContext?: WorkspaceFileScanContext;
|
||||
}) {
|
||||
let stdout: string;
|
||||
try {
|
||||
const result = await execFileAsync(
|
||||
"git",
|
||||
["-C", input.rootReal, "status", "--porcelain=v1", "-z", "--untracked-files=all"],
|
||||
{ maxBuffer: GIT_STATUS_MAX_BUFFER_BYTES },
|
||||
);
|
||||
const result = await workspaceGitOperationScheduler.run({
|
||||
workspacePath: input.rootReal,
|
||||
args: ["status", "--porcelain=v1", "-z", "--untracked-files=all"],
|
||||
operation: "workspace_file_browser.changed_files",
|
||||
fairnessKeys: input.scanContext?.fairnessKeys,
|
||||
signal: input.scanContext?.signal,
|
||||
maxStdoutBytes: GIT_STATUS_MAX_BUFFER_BYTES,
|
||||
maxStderrBytes: GIT_STATUS_MAX_BUFFER_BYTES,
|
||||
// The shared scheduler defaults this to ten seconds. This deliberately
|
||||
// trades a few seconds of freshness for protection from tab/refetch bursts.
|
||||
});
|
||||
stdout = result.stdout;
|
||||
} catch {
|
||||
} catch (error) {
|
||||
if (
|
||||
isWorkspaceGitScanError(error) &&
|
||||
error.code !== WORKSPACE_GIT_SCAN_ERROR_CODES.failed
|
||||
) {
|
||||
throw error;
|
||||
}
|
||||
return { unavailableReason: "changed_unavailable" as const };
|
||||
}
|
||||
|
||||
|
|
@ -1367,7 +1387,7 @@ export function workspaceFileResourceService(db: Db) {
|
|||
async function list(
|
||||
issueId: string,
|
||||
input: WorkspaceFileListQueryInput = {},
|
||||
opts: { issue?: IssueRow } = {},
|
||||
opts: { issue?: IssueRow; scanContext?: WorkspaceFileScanContext } = {},
|
||||
): Promise<WorkspaceFileListResponse> {
|
||||
const issue = opts.issue ?? await getIssue(issueId);
|
||||
const selector = input.workspace ?? "auto";
|
||||
|
|
@ -1442,7 +1462,14 @@ export function workspaceFileResourceService(db: Db) {
|
|||
}
|
||||
|
||||
if (mode === "changed") {
|
||||
const changed = await listChangedWorkspaceFiles({ candidate, rootReal, normalizedQuery, limit, offset });
|
||||
const changed = await listChangedWorkspaceFiles({
|
||||
candidate,
|
||||
rootReal,
|
||||
normalizedQuery,
|
||||
limit,
|
||||
offset,
|
||||
scanContext: opts.scanContext,
|
||||
});
|
||||
if ("unavailableReason" in changed) {
|
||||
const reason = changed.unavailableReason ?? "changed_unavailable";
|
||||
firstUnavailable ??= { candidate, reason };
|
||||
|
|
|
|||
|
|
@ -0,0 +1,424 @@
|
|||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
createWorkspaceGitOperationScheduler,
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES,
|
||||
WorkspaceGitScanError,
|
||||
workspaceGitSchedulerOptionsFromEnv,
|
||||
type WorkspaceGitRunner,
|
||||
} from "./workspace-git-operation-scheduler.js";
|
||||
|
||||
const tempPaths: string[] = [];
|
||||
|
||||
async function makeWorkspace(name = "workspace"): Promise<string> {
|
||||
const parent = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-git-scheduler-"));
|
||||
tempPaths.push(parent);
|
||||
const workspace = path.join(parent, name);
|
||||
await fs.mkdir(workspace, { recursive: true });
|
||||
return workspace;
|
||||
}
|
||||
|
||||
function deferred<T = void>() {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
let reject!: (reason?: unknown) => void;
|
||||
const promise = new Promise<T>((resolvePromise, rejectPromise) => {
|
||||
resolve = resolvePromise;
|
||||
reject = rejectPromise;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
function scanInput(workspacePath: string, suffix: string, fairnessKeys: string[] = []) {
|
||||
return {
|
||||
workspacePath,
|
||||
args: ["status", "--porcelain=v1", "-z", "--untracked-files=all", suffix],
|
||||
operation: "test.changed_files",
|
||||
fairnessKeys,
|
||||
cacheTtlMs: 0,
|
||||
};
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
vi.restoreAllMocks();
|
||||
await Promise.all(tempPaths.splice(0).map((tempPath) => fs.rm(tempPath, { recursive: true, force: true })));
|
||||
});
|
||||
|
||||
describe("WorkspaceGitOperationScheduler", () => {
|
||||
it("loads bounded process defaults and overrides from the environment", () => {
|
||||
expect(workspaceGitSchedulerOptionsFromEnv({})).toEqual({
|
||||
concurrency: 2,
|
||||
queueCapacity: 32,
|
||||
timeoutMs: 8_000,
|
||||
defaultCacheTtlMs: 10_000,
|
||||
});
|
||||
expect(workspaceGitSchedulerOptionsFromEnv({
|
||||
PAPERCLIP_WORKSPACE_GIT_SCAN_CONCURRENCY: "4",
|
||||
PAPERCLIP_WORKSPACE_GIT_SCAN_QUEUE_CAPACITY: "12",
|
||||
PAPERCLIP_WORKSPACE_GIT_SCAN_TIMEOUT_MS: "5000",
|
||||
PAPERCLIP_WORKSPACE_GIT_SCAN_CACHE_TTL_MS: "7000",
|
||||
})).toEqual({
|
||||
concurrency: 4,
|
||||
queueCapacity: 12,
|
||||
timeoutMs: 5_000,
|
||||
defaultCacheTtlMs: 7_000,
|
||||
});
|
||||
});
|
||||
|
||||
it("enforces process-wide concurrency across unrelated fairness keys", async () => {
|
||||
const workspace = await makeWorkspace();
|
||||
const releases: Array<() => void> = [];
|
||||
let active = 0;
|
||||
let peakActive = 0;
|
||||
const runner: WorkspaceGitRunner = () => new Promise((resolve) => {
|
||||
active += 1;
|
||||
peakActive = Math.max(peakActive, active);
|
||||
releases.push(() => {
|
||||
active -= 1;
|
||||
resolve({ stdout: "", stderr: "" });
|
||||
});
|
||||
});
|
||||
const scheduler = createWorkspaceGitOperationScheduler({ concurrency: 2, queueCapacity: 8, runner });
|
||||
|
||||
const requests = Array.from({ length: 6 }, (_, index) => scheduler.run(scanInput(
|
||||
workspace,
|
||||
String(index),
|
||||
[`company:${index}`, `actor:${index}`, `issue:${index}`],
|
||||
)));
|
||||
|
||||
await vi.waitFor(() => expect(scheduler.snapshot()).toMatchObject({ activeCount: 2, queuedCount: 4 }));
|
||||
for (let completed = 0; completed < requests.length; completed += 1) {
|
||||
await vi.waitFor(() => expect(releases.length).toBeGreaterThan(0));
|
||||
releases.shift()?.();
|
||||
}
|
||||
await Promise.all(requests);
|
||||
|
||||
expect(peakActive).toBe(2);
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, queuedCount: 0, inFlightCount: 0 });
|
||||
});
|
||||
|
||||
it("honors per-operation deadlines and keeps different execution bounds out of one flight", async () => {
|
||||
const workspace = await makeWorkspace();
|
||||
const observedTimeouts: number[] = [];
|
||||
const runner: WorkspaceGitRunner = async (input) => {
|
||||
observedTimeouts.push(input.timeoutMs);
|
||||
return { stdout: "", stderr: "" };
|
||||
};
|
||||
const scheduler = createWorkspaceGitOperationScheduler({
|
||||
concurrency: 2,
|
||||
timeoutMs: 8_000,
|
||||
runner,
|
||||
});
|
||||
const input = scanInput(workspace, "same");
|
||||
|
||||
await Promise.all([
|
||||
scheduler.run({ ...input, timeoutMs: 10_000, maxStdoutBytes: 1024 }),
|
||||
scheduler.run({ ...input, timeoutMs: 12_000, maxStdoutBytes: 1024 }),
|
||||
]);
|
||||
|
||||
expect(observedTimeouts.sort((a, b) => a - b)).toEqual([10_000, 12_000]);
|
||||
expect(scheduler.snapshot().totals.singleFlightJoins).toBe(0);
|
||||
});
|
||||
|
||||
it("bounds the queue and fails excess work immediately with a typed retryable error", async () => {
|
||||
const workspace = await makeWorkspace();
|
||||
const releases: Array<() => void> = [];
|
||||
const runner: WorkspaceGitRunner = () => new Promise((resolve) => {
|
||||
releases.push(() => resolve({ stdout: "", stderr: "" }));
|
||||
});
|
||||
const scheduler = createWorkspaceGitOperationScheduler({ concurrency: 1, queueCapacity: 1, runner });
|
||||
|
||||
const active = scheduler.run(scanInput(workspace, "active"));
|
||||
await vi.waitFor(() => expect(scheduler.snapshot().activeCount).toBe(1));
|
||||
const queued = scheduler.run(scanInput(workspace, "queued"));
|
||||
await vi.waitFor(() => expect(scheduler.snapshot().queuedCount).toBe(1));
|
||||
|
||||
await expect(scheduler.run(scanInput(workspace, "rejected"))).rejects.toMatchObject({
|
||||
status: 503,
|
||||
code: WORKSPACE_GIT_SCAN_ERROR_CODES.saturated,
|
||||
details: expect.objectContaining({ retryable: true }),
|
||||
});
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 1, queuedCount: 1 });
|
||||
|
||||
releases.shift()?.();
|
||||
await vi.waitFor(() => expect(releases).toHaveLength(1));
|
||||
releases.shift()?.();
|
||||
await Promise.all([active, queued]);
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, queuedCount: 0, inFlightCount: 0 });
|
||||
});
|
||||
|
||||
it("coalesces the same canonical key and cleans single-flight state after success and failure", async () => {
|
||||
const workspace = await makeWorkspace();
|
||||
const alias = `${workspace}-alias`;
|
||||
await fs.symlink(workspace, alias, "dir");
|
||||
tempPaths.push(alias);
|
||||
const gate = deferred<void>();
|
||||
let calls = 0;
|
||||
let shouldFail = false;
|
||||
const runner: WorkspaceGitRunner = async () => {
|
||||
calls += 1;
|
||||
await gate.promise;
|
||||
if (shouldFail) throw new Error("synthetic failure");
|
||||
return { stdout: "shared", stderr: "" };
|
||||
};
|
||||
const scheduler = createWorkspaceGitOperationScheduler({ runner, defaultCacheTtlMs: 0 });
|
||||
|
||||
const first = scheduler.run(scanInput(workspace, "same"));
|
||||
const joined = scheduler.run(scanInput(alias, "same"));
|
||||
await vi.waitFor(() => expect(scheduler.snapshot().totals.singleFlightJoins).toBe(1));
|
||||
expect(calls).toBe(1);
|
||||
gate.resolve();
|
||||
await expect(Promise.all([first, joined])).resolves.toEqual([
|
||||
expect.objectContaining({ stdout: "shared", singleFlightJoined: false }),
|
||||
expect.objectContaining({ stdout: "shared", singleFlightJoined: true }),
|
||||
]);
|
||||
|
||||
shouldFail = true;
|
||||
await expect(scheduler.run(scanInput(workspace, "failure"))).rejects.toMatchObject({
|
||||
code: WORKSPACE_GIT_SCAN_ERROR_CODES.failed,
|
||||
});
|
||||
shouldFail = false;
|
||||
await expect(scheduler.run(scanInput(workspace, "failure"))).resolves.toMatchObject({ stdout: "shared" });
|
||||
expect(calls).toBe(3);
|
||||
expect(scheduler.snapshot().inFlightCount).toBe(0);
|
||||
});
|
||||
|
||||
it("serves cached results until TTL expiry and evicts least-recently-used entries", async () => {
|
||||
const firstWorkspace = await makeWorkspace("same-name");
|
||||
const secondWorkspace = await makeWorkspace("same-name");
|
||||
const thirdWorkspace = await makeWorkspace("same-name");
|
||||
let now = 1_000;
|
||||
let calls = 0;
|
||||
const runner: WorkspaceGitRunner = async ({ canonicalWorkspacePath }) => {
|
||||
calls += 1;
|
||||
return { stdout: `${canonicalWorkspacePath}:${calls}`, stderr: "" };
|
||||
};
|
||||
const scheduler = createWorkspaceGitOperationScheduler({
|
||||
runner,
|
||||
now: () => now,
|
||||
defaultCacheTtlMs: 100,
|
||||
maxCacheEntries: 2,
|
||||
});
|
||||
const cacheable = (workspacePath: string) => ({
|
||||
...scanInput(workspacePath, "same"),
|
||||
cacheTtlMs: 100,
|
||||
});
|
||||
|
||||
const first = await scheduler.run(cacheable(firstWorkspace));
|
||||
const cacheHit = await scheduler.run(cacheable(firstWorkspace));
|
||||
expect(cacheHit).toMatchObject({ stdout: first.stdout, cacheHit: true });
|
||||
expect(calls).toBe(1);
|
||||
|
||||
const bypass = await scheduler.run({ ...cacheable(firstWorkspace), cacheTtlMs: 0 });
|
||||
expect(bypass.cacheHit).toBe(false);
|
||||
expect(calls).toBe(2);
|
||||
|
||||
now += 101;
|
||||
await scheduler.run(cacheable(firstWorkspace));
|
||||
expect(calls).toBe(3);
|
||||
await scheduler.run(cacheable(secondWorkspace));
|
||||
// Touch the first entry, then force the second (the LRU) out.
|
||||
await scheduler.run(cacheable(firstWorkspace));
|
||||
await scheduler.run(cacheable(thirdWorkspace));
|
||||
expect(scheduler.snapshot().cacheEntryCount).toBe(2);
|
||||
await scheduler.run(cacheable(secondWorkspace));
|
||||
expect(calls).toBe(6);
|
||||
});
|
||||
|
||||
it("keeps identical display paths in different canonical workspaces isolated", async () => {
|
||||
const firstWorkspace = await makeWorkspace("repo");
|
||||
const secondWorkspace = await makeWorkspace("repo");
|
||||
let calls = 0;
|
||||
const scheduler = createWorkspaceGitOperationScheduler({
|
||||
runner: async ({ canonicalWorkspacePath }) => {
|
||||
calls += 1;
|
||||
return { stdout: canonicalWorkspacePath, stderr: "" };
|
||||
},
|
||||
defaultCacheTtlMs: 1_000,
|
||||
});
|
||||
|
||||
const [first, second] = await Promise.all([
|
||||
scheduler.run({ ...scanInput(firstWorkspace, "same"), cacheTtlMs: 1_000 }),
|
||||
scheduler.run({ ...scanInput(secondWorkspace, "same"), cacheTtlMs: 1_000 }),
|
||||
]);
|
||||
|
||||
expect(first.stdout).not.toBe(second.stdout);
|
||||
expect(first.workspaceHash).not.toBe(second.workspaceHash);
|
||||
expect(calls).toBe(2);
|
||||
});
|
||||
|
||||
it("does not let a repeatedly served fairness group monopolize the next slot", async () => {
|
||||
const workspace = await makeWorkspace();
|
||||
const order: string[] = [];
|
||||
const releases: Array<() => void> = [];
|
||||
const runner: WorkspaceGitRunner = ({ args }) => new Promise((resolve) => {
|
||||
const name = args.at(-1)!;
|
||||
order.push(name);
|
||||
releases.push(() => resolve({ stdout: name, stderr: "" }));
|
||||
});
|
||||
const scheduler = createWorkspaceGitOperationScheduler({ concurrency: 1, queueCapacity: 4, runner });
|
||||
|
||||
const firstA = scheduler.run(scanInput(workspace, "a-1", ["company:a", "actor:a"]));
|
||||
await vi.waitFor(() => expect(order).toEqual(["a-1"]));
|
||||
const secondA = scheduler.run(scanInput(workspace, "a-2", ["company:a", "actor:a"]));
|
||||
const firstB = scheduler.run(scanInput(workspace, "b-1", ["company:b", "actor:b"]));
|
||||
await vi.waitFor(() => expect(scheduler.snapshot().queuedCount).toBe(2));
|
||||
releases.shift()?.();
|
||||
await vi.waitFor(() => expect(order).toEqual(["a-1", "b-1"]));
|
||||
releases.shift()?.();
|
||||
await vi.waitFor(() => expect(order).toEqual(["a-1", "b-1", "a-2"]));
|
||||
releases.shift()?.();
|
||||
|
||||
await Promise.all([firstA, secondA, firstB]);
|
||||
});
|
||||
|
||||
it("keeps a shared scan alive for remaining waiters and cancels it after the last disconnect", async () => {
|
||||
const workspace = await makeWorkspace();
|
||||
let underlyingAborted = false;
|
||||
const runner: WorkspaceGitRunner = ({ signal, canonicalWorkspacePath }) => new Promise((_resolve, reject) => {
|
||||
signal.addEventListener("abort", () => {
|
||||
underlyingAborted = true;
|
||||
reject(new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled,
|
||||
"cancelled",
|
||||
{ canonicalWorkspacePath },
|
||||
));
|
||||
}, { once: true });
|
||||
});
|
||||
const scheduler = createWorkspaceGitOperationScheduler({ runner });
|
||||
const firstController = new AbortController();
|
||||
const secondController = new AbortController();
|
||||
const first = scheduler.run({ ...scanInput(workspace, "same"), signal: firstController.signal }).catch((error) => error);
|
||||
const second = scheduler.run({ ...scanInput(workspace, "same"), signal: secondController.signal }).catch((error) => error);
|
||||
await vi.waitFor(() => expect(scheduler.snapshot().totals.singleFlightJoins).toBe(1));
|
||||
|
||||
firstController.abort();
|
||||
await expect(first).resolves.toMatchObject({ code: WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled });
|
||||
expect(underlyingAborted).toBe(false);
|
||||
expect(scheduler.snapshot().activeCount).toBe(1);
|
||||
|
||||
secondController.abort();
|
||||
await expect(second).resolves.toMatchObject({ code: WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled });
|
||||
await vi.waitFor(() => expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, inFlightCount: 0 }));
|
||||
expect(underlyingAborted).toBe(true);
|
||||
});
|
||||
|
||||
it("removes an abandoned queued scan without consuming a scheduler slot", async () => {
|
||||
const workspace = await makeWorkspace();
|
||||
const gate = deferred<void>();
|
||||
const scheduler = createWorkspaceGitOperationScheduler({
|
||||
concurrency: 1,
|
||||
queueCapacity: 2,
|
||||
runner: async () => {
|
||||
await gate.promise;
|
||||
return { stdout: "", stderr: "" };
|
||||
},
|
||||
});
|
||||
const active = scheduler.run(scanInput(workspace, "active"));
|
||||
await vi.waitFor(() => expect(scheduler.snapshot().activeCount).toBe(1));
|
||||
const controller = new AbortController();
|
||||
const queued = scheduler.run({
|
||||
...scanInput(workspace, "queued"),
|
||||
signal: controller.signal,
|
||||
}).catch((error) => error);
|
||||
await vi.waitFor(() => expect(scheduler.snapshot().queuedCount).toBe(1));
|
||||
|
||||
controller.abort();
|
||||
await expect(queued).resolves.toMatchObject({ code: WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled });
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 1, queuedCount: 0, inFlightCount: 1 });
|
||||
|
||||
gate.resolve();
|
||||
await active;
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, queuedCount: 0, inFlightCount: 0 });
|
||||
});
|
||||
|
||||
it("kills a hung subprocess after the hard timeout and releases the slot for the next scan", async () => {
|
||||
const workspace = await makeWorkspace();
|
||||
const scriptPath = path.join(path.dirname(workspace), "fake-git.mjs");
|
||||
const pidPath = path.join(path.dirname(workspace), "fake-git.pid");
|
||||
await fs.writeFile(scriptPath, [
|
||||
'import fs from "node:fs";',
|
||||
'fs.writeFileSync(process.env.PAPERCLIP_FAKE_GIT_PID_PATH, String(process.pid));',
|
||||
'if (process.argv.includes("hang")) {',
|
||||
' process.on("SIGTERM", () => {});',
|
||||
' setInterval(() => {}, 1000);',
|
||||
'} else {',
|
||||
' process.stdout.write("ok");',
|
||||
'}',
|
||||
].join("\n"), "utf8");
|
||||
const scheduler = createWorkspaceGitOperationScheduler({
|
||||
concurrency: 1,
|
||||
timeoutMs: 500,
|
||||
killGraceMs: 50,
|
||||
gitBinary: process.execPath,
|
||||
gitArgsPrefix: [scriptPath],
|
||||
});
|
||||
const env = {
|
||||
...(process.env.SystemRoot ? { SystemRoot: process.env.SystemRoot } : {}),
|
||||
PAPERCLIP_FAKE_GIT_PID_PATH: pidPath,
|
||||
};
|
||||
|
||||
await expect(scheduler.run({ ...scanInput(workspace, "hang"), env })).rejects.toMatchObject({
|
||||
status: 504,
|
||||
code: WORKSPACE_GIT_SCAN_ERROR_CODES.timeout,
|
||||
});
|
||||
const killedPid = Number(await fs.readFile(pidPath, "utf8"));
|
||||
expect(() => process.kill(killedPid, 0)).toThrow();
|
||||
const outputScheduler = createWorkspaceGitOperationScheduler({
|
||||
concurrency: 1,
|
||||
timeoutMs: 1_000,
|
||||
killGraceMs: 50,
|
||||
gitBinary: process.execPath,
|
||||
gitArgsPrefix: ["-e", 'process.stdout.write("x".repeat(65536)); setInterval(() => {}, 1000);'],
|
||||
});
|
||||
await expect(outputScheduler.run({
|
||||
...scanInput(workspace, "flood"),
|
||||
env,
|
||||
maxStdoutBytes: 32,
|
||||
})).rejects.toMatchObject({
|
||||
status: 503,
|
||||
code: WORKSPACE_GIT_SCAN_ERROR_CODES.outputLimit,
|
||||
});
|
||||
expect(outputScheduler.snapshot()).toMatchObject({ activeCount: 0, inFlightCount: 0 });
|
||||
await expect(scheduler.run({ ...scanInput(workspace, "ok"), env })).resolves.toMatchObject({
|
||||
cacheHit: false,
|
||||
singleFlightJoined: false,
|
||||
});
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, queuedCount: 0, inFlightCount: 0 });
|
||||
});
|
||||
|
||||
it("coalesces 500 requests over two repositories into two bounded scans", async () => {
|
||||
const firstWorkspace = await makeWorkspace("repo-a");
|
||||
const secondWorkspace = await makeWorkspace("repo-b");
|
||||
const gate = deferred<void>();
|
||||
let active = 0;
|
||||
let peakActive = 0;
|
||||
let calls = 0;
|
||||
const scheduler = createWorkspaceGitOperationScheduler({
|
||||
concurrency: 2,
|
||||
queueCapacity: 4,
|
||||
runner: async () => {
|
||||
calls += 1;
|
||||
active += 1;
|
||||
peakActive = Math.max(peakActive, active);
|
||||
await gate.promise;
|
||||
active -= 1;
|
||||
return { stdout: "", stderr: "" };
|
||||
},
|
||||
});
|
||||
const requests = Array.from({ length: 500 }, (_, index) => scheduler.run(scanInput(
|
||||
index % 2 === 0 ? firstWorkspace : secondWorkspace,
|
||||
"same",
|
||||
[`company:${index % 17}`, `actor:${index % 73}`, `issue:${index}`],
|
||||
)));
|
||||
|
||||
await vi.waitFor(() => expect(scheduler.snapshot().totals.singleFlightJoins).toBe(498));
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 2, queuedCount: 0, inFlightCount: 2 });
|
||||
gate.resolve();
|
||||
await Promise.all(requests);
|
||||
|
||||
expect({ calls, peakActive }).toEqual({ calls: 2, peakActive: 2 });
|
||||
expect(scheduler.snapshot()).toMatchObject({ activeCount: 0, queuedCount: 0, inFlightCount: 0 });
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,891 @@
|
|||
import { spawn, type ChildProcess } from "node:child_process";
|
||||
import { createHash } from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import { setExpensiveWorkspaceGitExecutor } from "@paperclipai/adapter-utils/git-workspace-sync";
|
||||
import { HttpError } from "../errors.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
|
||||
export const WORKSPACE_GIT_SCAN_ERROR_CODES = {
|
||||
saturated: "workspace_git_scan_saturated",
|
||||
timeout: "workspace_git_scan_timeout",
|
||||
cancelled: "workspace_git_scan_cancelled",
|
||||
outputLimit: "workspace_git_scan_output_limit",
|
||||
failed: "workspace_git_scan_failed",
|
||||
} as const;
|
||||
|
||||
export type WorkspaceGitScanErrorCode =
|
||||
(typeof WORKSPACE_GIT_SCAN_ERROR_CODES)[keyof typeof WORKSPACE_GIT_SCAN_ERROR_CODES];
|
||||
|
||||
export class WorkspaceGitScanError extends HttpError {
|
||||
readonly code: WorkspaceGitScanErrorCode;
|
||||
|
||||
constructor(
|
||||
code: WorkspaceGitScanErrorCode,
|
||||
message: string,
|
||||
details: Record<string, unknown> = {},
|
||||
) {
|
||||
const status = code === WORKSPACE_GIT_SCAN_ERROR_CODES.timeout
|
||||
? 504
|
||||
: code === WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled
|
||||
? 499
|
||||
: 503;
|
||||
super(status, message, {
|
||||
code,
|
||||
retryable: code !== WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled,
|
||||
...details,
|
||||
});
|
||||
this.name = "WorkspaceGitScanError";
|
||||
this.code = code;
|
||||
}
|
||||
}
|
||||
|
||||
export function isWorkspaceGitScanError(error: unknown): error is WorkspaceGitScanError {
|
||||
return error instanceof WorkspaceGitScanError;
|
||||
}
|
||||
|
||||
export interface WorkspaceGitScanResult {
|
||||
stdout: string;
|
||||
stderr: string;
|
||||
canonicalWorkspacePath: string;
|
||||
workspaceHash: string;
|
||||
cacheHit: boolean;
|
||||
singleFlightJoined: boolean;
|
||||
}
|
||||
|
||||
export interface WorkspaceGitScanInput {
|
||||
workspacePath: string;
|
||||
args: readonly string[];
|
||||
operation: string;
|
||||
/**
|
||||
* Stable, non-secret scheduling dimensions. Jobs sharing any recently-served
|
||||
* dimension are deprioritized, so changing issue ids cannot bypass fairness
|
||||
* for the same company, actor, or repository.
|
||||
*/
|
||||
fairnessKeys?: readonly string[];
|
||||
signal?: AbortSignal;
|
||||
/** Successful-result cache duration. Use zero for correctness-sensitive guards. */
|
||||
cacheTtlMs?: number;
|
||||
/** Per-operation wall-clock deadline. Defaults to the process-wide setting. */
|
||||
timeoutMs?: number;
|
||||
env?: NodeJS.ProcessEnv;
|
||||
maxStdoutBytes?: number;
|
||||
maxStderrBytes?: number;
|
||||
}
|
||||
|
||||
export interface WorkspaceGitSchedulerSnapshot {
|
||||
activeCount: number;
|
||||
queuedCount: number;
|
||||
inFlightCount: number;
|
||||
cacheEntryCount: number;
|
||||
cacheBytes: number;
|
||||
totals: {
|
||||
started: number;
|
||||
succeeded: number;
|
||||
failed: number;
|
||||
timedOut: number;
|
||||
cancelled: number;
|
||||
saturated: number;
|
||||
cacheHits: number;
|
||||
singleFlightJoins: number;
|
||||
};
|
||||
}
|
||||
|
||||
export interface WorkspaceGitRunnerInput {
|
||||
canonicalWorkspacePath: string;
|
||||
args: readonly string[];
|
||||
env?: NodeJS.ProcessEnv;
|
||||
signal: AbortSignal;
|
||||
timeoutMs: number;
|
||||
killGraceMs: number;
|
||||
maxStdoutBytes: number;
|
||||
maxStderrBytes: number;
|
||||
}
|
||||
|
||||
export interface WorkspaceGitRunnerResult {
|
||||
stdout: string;
|
||||
stderr: string;
|
||||
}
|
||||
|
||||
export type WorkspaceGitRunner = (
|
||||
input: WorkspaceGitRunnerInput,
|
||||
) => Promise<WorkspaceGitRunnerResult>;
|
||||
|
||||
export interface WorkspaceGitOperationSchedulerOptions {
|
||||
concurrency?: number;
|
||||
queueCapacity?: number;
|
||||
timeoutMs?: number;
|
||||
killGraceMs?: number;
|
||||
defaultCacheTtlMs?: number;
|
||||
maxCacheEntries?: number;
|
||||
maxCacheBytes?: number;
|
||||
maxStdoutBytes?: number;
|
||||
maxStderrBytes?: number;
|
||||
warningIntervalMs?: number;
|
||||
gitBinary?: string;
|
||||
gitArgsPrefix?: readonly string[];
|
||||
runner?: WorkspaceGitRunner;
|
||||
now?: () => number;
|
||||
}
|
||||
|
||||
interface Waiter {
|
||||
id: symbol;
|
||||
signal?: AbortSignal;
|
||||
onAbort?: () => void;
|
||||
resolve: (result: WorkspaceGitScanResult) => void;
|
||||
reject: (error: unknown) => void;
|
||||
joined: boolean;
|
||||
}
|
||||
|
||||
interface PendingScan {
|
||||
key: string;
|
||||
operation: string;
|
||||
canonicalWorkspacePath: string;
|
||||
workspaceHash: string;
|
||||
args: readonly string[];
|
||||
fairnessKeys: readonly string[];
|
||||
env?: NodeJS.ProcessEnv;
|
||||
timeoutMs: number;
|
||||
maxStdoutBytes: number;
|
||||
maxStderrBytes: number;
|
||||
cacheTtlMs: number;
|
||||
enqueuedAt: number;
|
||||
state: "queued" | "running";
|
||||
controller: AbortController;
|
||||
waiters: Map<symbol, Waiter>;
|
||||
joinCount: number;
|
||||
}
|
||||
|
||||
interface CacheEntry {
|
||||
stdout: string;
|
||||
stderr: string;
|
||||
canonicalWorkspacePath: string;
|
||||
workspaceHash: string;
|
||||
expiresAt: number;
|
||||
bytes: number;
|
||||
}
|
||||
|
||||
interface WarningBucket {
|
||||
lastLoggedAt: number;
|
||||
suppressed: number;
|
||||
}
|
||||
|
||||
const DEFAULT_CONCURRENCY = 2;
|
||||
const DEFAULT_QUEUE_CAPACITY = 32;
|
||||
const DEFAULT_TIMEOUT_MS = 8_000;
|
||||
const DEFAULT_KILL_GRACE_MS = 250;
|
||||
const DEFAULT_CACHE_TTL_MS = 10_000;
|
||||
const DEFAULT_CACHE_ENTRIES = 64;
|
||||
const DEFAULT_CACHE_BYTES = 8 * 1024 * 1024;
|
||||
const DEFAULT_OUTPUT_BYTES = 1024 * 1024;
|
||||
const DEFAULT_WARNING_INTERVAL_MS = 10_000;
|
||||
|
||||
function clampInteger(value: number | undefined, fallback: number, min: number, max: number): number {
|
||||
if (!Number.isFinite(value)) return fallback;
|
||||
return Math.min(max, Math.max(min, Math.floor(value!)));
|
||||
}
|
||||
|
||||
function envInteger(
|
||||
env: NodeJS.ProcessEnv,
|
||||
key: string,
|
||||
fallback: number,
|
||||
min: number,
|
||||
max: number,
|
||||
): number {
|
||||
const raw = env[key]?.trim();
|
||||
if (!raw) return fallback;
|
||||
const parsed = Number(raw);
|
||||
return clampInteger(parsed, fallback, min, max);
|
||||
}
|
||||
|
||||
export function workspaceGitSchedulerOptionsFromEnv(
|
||||
env: NodeJS.ProcessEnv = process.env,
|
||||
): Required<Pick<
|
||||
WorkspaceGitOperationSchedulerOptions,
|
||||
"concurrency" | "queueCapacity" | "timeoutMs" | "defaultCacheTtlMs"
|
||||
>> {
|
||||
return {
|
||||
concurrency: envInteger(env, "PAPERCLIP_WORKSPACE_GIT_SCAN_CONCURRENCY", DEFAULT_CONCURRENCY, 1, 16),
|
||||
queueCapacity: envInteger(env, "PAPERCLIP_WORKSPACE_GIT_SCAN_QUEUE_CAPACITY", DEFAULT_QUEUE_CAPACITY, 0, 1_024),
|
||||
timeoutMs: envInteger(env, "PAPERCLIP_WORKSPACE_GIT_SCAN_TIMEOUT_MS", DEFAULT_TIMEOUT_MS, 100, 120_000),
|
||||
defaultCacheTtlMs: envInteger(env, "PAPERCLIP_WORKSPACE_GIT_SCAN_CACHE_TTL_MS", DEFAULT_CACHE_TTL_MS, 0, 60_000),
|
||||
};
|
||||
}
|
||||
|
||||
function workspaceIdentity(canonicalWorkspacePath: string): string {
|
||||
return createHash("sha256").update(canonicalWorkspacePath).digest("hex").slice(0, 16);
|
||||
}
|
||||
|
||||
function scanKey(input: {
|
||||
canonicalWorkspacePath: string;
|
||||
args: readonly string[];
|
||||
env?: NodeJS.ProcessEnv;
|
||||
timeoutMs: number;
|
||||
maxStdoutBytes: number;
|
||||
maxStderrBytes: number;
|
||||
}): string {
|
||||
// These variables can change status semantics. Hash values so neither keys nor
|
||||
// telemetry expose credentials or private config contents.
|
||||
const effectiveEnv = input.env ?? process.env;
|
||||
const semanticEnv = [
|
||||
"GIT_CONFIG_COUNT",
|
||||
"GIT_CONFIG_PARAMETERS",
|
||||
"GIT_DIR",
|
||||
"GIT_INDEX_FILE",
|
||||
"GIT_OPTIONAL_LOCKS",
|
||||
"GIT_WORK_TREE",
|
||||
].map((key) => [key, effectiveEnv[key] ?? null]);
|
||||
return createHash("sha256")
|
||||
.update(JSON.stringify({
|
||||
workspacePath: input.canonicalWorkspacePath,
|
||||
args: input.args,
|
||||
semanticEnv,
|
||||
timeoutMs: input.timeoutMs,
|
||||
maxStdoutBytes: input.maxStdoutBytes,
|
||||
maxStderrBytes: input.maxStderrBytes,
|
||||
}))
|
||||
.digest("hex");
|
||||
}
|
||||
|
||||
function abortError(workspaceHash: string): WorkspaceGitScanError {
|
||||
return new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled,
|
||||
"Workspace Git scan was cancelled",
|
||||
{ workspaceHash },
|
||||
);
|
||||
}
|
||||
|
||||
function signalChild(child: ChildProcess, signal: NodeJS.Signals): void {
|
||||
if (child.exitCode !== null || child.signalCode !== null) return;
|
||||
if (process.platform !== "win32" && child.pid) {
|
||||
try {
|
||||
process.kill(-child.pid, signal);
|
||||
return;
|
||||
} catch {
|
||||
// Fall back to the direct child if the process group already disappeared.
|
||||
}
|
||||
}
|
||||
try {
|
||||
child.kill(signal);
|
||||
} catch {
|
||||
// The close/error handler owns settlement; an already-dead child is benign.
|
||||
}
|
||||
}
|
||||
|
||||
function createSpawnRunner(input: {
|
||||
gitBinary: string;
|
||||
gitArgsPrefix: readonly string[];
|
||||
}): WorkspaceGitRunner {
|
||||
return (runInput) => new Promise<WorkspaceGitRunnerResult>((resolve, reject) => {
|
||||
if (runInput.signal.aborted) {
|
||||
reject(abortError(workspaceIdentity(runInput.canonicalWorkspacePath)));
|
||||
return;
|
||||
}
|
||||
|
||||
const child = spawn(
|
||||
input.gitBinary,
|
||||
[...input.gitArgsPrefix, "-C", runInput.canonicalWorkspacePath, ...runInput.args],
|
||||
{
|
||||
cwd: runInput.canonicalWorkspacePath,
|
||||
env: runInput.env ?? process.env,
|
||||
stdio: ["ignore", "pipe", "pipe"],
|
||||
detached: process.platform !== "win32",
|
||||
windowsHide: true,
|
||||
},
|
||||
);
|
||||
const stdoutChunks: Buffer[] = [];
|
||||
const stderrChunks: Buffer[] = [];
|
||||
let stdoutBytes = 0;
|
||||
let stderrBytes = 0;
|
||||
let termination: "timeout" | "cancelled" | "output_limit" | null = null;
|
||||
let spawnError: Error | null = null;
|
||||
let settled = false;
|
||||
|
||||
const terminate = (reason: NonNullable<typeof termination>) => {
|
||||
if (termination) return;
|
||||
termination = reason;
|
||||
signalChild(child, "SIGTERM");
|
||||
killTimer = setTimeout(() => signalChild(child, "SIGKILL"), runInput.killGraceMs);
|
||||
killTimer.unref?.();
|
||||
};
|
||||
|
||||
const append = (
|
||||
chunk: Buffer | string,
|
||||
chunks: Buffer[],
|
||||
currentBytes: number,
|
||||
maxBytes: number,
|
||||
): number => {
|
||||
const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
|
||||
const remaining = Math.max(0, maxBytes - currentBytes);
|
||||
if (remaining > 0) chunks.push(buffer.subarray(0, remaining));
|
||||
const nextBytes = currentBytes + buffer.length;
|
||||
if (nextBytes > maxBytes) terminate("output_limit");
|
||||
return nextBytes;
|
||||
};
|
||||
|
||||
const onAbort = () => terminate("cancelled");
|
||||
runInput.signal.addEventListener("abort", onAbort, { once: true });
|
||||
child.stdout?.on("data", (chunk) => {
|
||||
stdoutBytes = append(chunk, stdoutChunks, stdoutBytes, runInput.maxStdoutBytes);
|
||||
});
|
||||
child.stderr?.on("data", (chunk) => {
|
||||
stderrBytes = append(chunk, stderrChunks, stderrBytes, runInput.maxStderrBytes);
|
||||
});
|
||||
child.once("error", (error) => {
|
||||
spawnError = error;
|
||||
});
|
||||
|
||||
const timeoutTimer = setTimeout(() => terminate("timeout"), runInput.timeoutMs);
|
||||
timeoutTimer.unref?.();
|
||||
let killTimer: NodeJS.Timeout | null = null;
|
||||
|
||||
child.once("close", (code, childSignal) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timeoutTimer);
|
||||
if (killTimer) clearTimeout(killTimer);
|
||||
runInput.signal.removeEventListener("abort", onAbort);
|
||||
const stdout = Buffer.concat(stdoutChunks).toString("utf8");
|
||||
const stderr = Buffer.concat(stderrChunks).toString("utf8");
|
||||
const workspaceHash = workspaceIdentity(runInput.canonicalWorkspacePath);
|
||||
|
||||
if (termination === "timeout") {
|
||||
reject(new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.timeout,
|
||||
`Workspace Git scan timed out after ${runInput.timeoutMs}ms`,
|
||||
{ workspaceHash, timeoutMs: runInput.timeoutMs },
|
||||
));
|
||||
return;
|
||||
}
|
||||
if (termination === "cancelled") {
|
||||
reject(abortError(workspaceHash));
|
||||
return;
|
||||
}
|
||||
if (termination === "output_limit") {
|
||||
reject(new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.outputLimit,
|
||||
"Workspace Git scan exceeded its output limit",
|
||||
{
|
||||
workspaceHash,
|
||||
stdoutBytes,
|
||||
stderrBytes,
|
||||
maxStdoutBytes: runInput.maxStdoutBytes,
|
||||
maxStderrBytes: runInput.maxStderrBytes,
|
||||
},
|
||||
));
|
||||
return;
|
||||
}
|
||||
if (spawnError) {
|
||||
reject(new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.failed,
|
||||
"Workspace Git scan could not start",
|
||||
{ workspaceHash, cause: spawnError.message },
|
||||
));
|
||||
return;
|
||||
}
|
||||
if (code !== 0) {
|
||||
reject(new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.failed,
|
||||
"Workspace Git scan failed",
|
||||
{
|
||||
workspaceHash,
|
||||
exitCode: code,
|
||||
signal: childSignal,
|
||||
// Keep the diagnostic bounded; callers never receive raw paths in telemetry.
|
||||
stderr: stderr.trim().slice(0, 1_000),
|
||||
},
|
||||
));
|
||||
return;
|
||||
}
|
||||
resolve({ stdout, stderr });
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
export class WorkspaceGitOperationScheduler {
|
||||
private readonly concurrency: number;
|
||||
private readonly queueCapacity: number;
|
||||
private readonly timeoutMs: number;
|
||||
private readonly killGraceMs: number;
|
||||
private readonly defaultCacheTtlMs: number;
|
||||
private readonly maxCacheEntries: number;
|
||||
private readonly maxCacheBytes: number;
|
||||
private readonly maxStdoutBytes: number;
|
||||
private readonly maxStderrBytes: number;
|
||||
private readonly warningIntervalMs: number;
|
||||
private readonly runner: WorkspaceGitRunner;
|
||||
private readonly now: () => number;
|
||||
private readonly inFlight = new Map<string, PendingScan>();
|
||||
private readonly queue: PendingScan[] = [];
|
||||
private readonly cache = new Map<string, CacheEntry>();
|
||||
private readonly lastServedByFairnessKey = new Map<string, number>();
|
||||
private readonly warningBuckets = new Map<string, WarningBucket>();
|
||||
private cacheBytes = 0;
|
||||
private activeCount = 0;
|
||||
private serviceSequence = 0;
|
||||
private readonly totals = {
|
||||
started: 0,
|
||||
succeeded: 0,
|
||||
failed: 0,
|
||||
timedOut: 0,
|
||||
cancelled: 0,
|
||||
saturated: 0,
|
||||
cacheHits: 0,
|
||||
singleFlightJoins: 0,
|
||||
};
|
||||
|
||||
constructor(options: WorkspaceGitOperationSchedulerOptions = {}) {
|
||||
this.concurrency = clampInteger(options.concurrency, DEFAULT_CONCURRENCY, 1, 16);
|
||||
this.queueCapacity = clampInteger(options.queueCapacity, DEFAULT_QUEUE_CAPACITY, 0, 1_024);
|
||||
this.timeoutMs = clampInteger(options.timeoutMs, DEFAULT_TIMEOUT_MS, 1, 120_000);
|
||||
this.killGraceMs = clampInteger(options.killGraceMs, DEFAULT_KILL_GRACE_MS, 1, 10_000);
|
||||
this.defaultCacheTtlMs = clampInteger(options.defaultCacheTtlMs, DEFAULT_CACHE_TTL_MS, 0, 60_000);
|
||||
this.maxCacheEntries = clampInteger(options.maxCacheEntries, DEFAULT_CACHE_ENTRIES, 0, 10_000);
|
||||
this.maxCacheBytes = clampInteger(options.maxCacheBytes, DEFAULT_CACHE_BYTES, 0, 1024 * 1024 * 1024);
|
||||
this.maxStdoutBytes = clampInteger(options.maxStdoutBytes, DEFAULT_OUTPUT_BYTES, 1, 128 * 1024 * 1024);
|
||||
this.maxStderrBytes = clampInteger(options.maxStderrBytes, DEFAULT_OUTPUT_BYTES, 1, 128 * 1024 * 1024);
|
||||
this.warningIntervalMs = clampInteger(options.warningIntervalMs, DEFAULT_WARNING_INTERVAL_MS, 1, 60 * 60_000);
|
||||
this.now = options.now ?? Date.now;
|
||||
this.runner = options.runner ?? createSpawnRunner({
|
||||
gitBinary: options.gitBinary ?? "git",
|
||||
gitArgsPrefix: options.gitArgsPrefix ?? [],
|
||||
});
|
||||
}
|
||||
|
||||
snapshot(): WorkspaceGitSchedulerSnapshot {
|
||||
this.pruneExpiredCache();
|
||||
return {
|
||||
activeCount: this.activeCount,
|
||||
queuedCount: this.queue.length,
|
||||
inFlightCount: this.inFlight.size,
|
||||
cacheEntryCount: this.cache.size,
|
||||
cacheBytes: this.cacheBytes,
|
||||
totals: { ...this.totals },
|
||||
};
|
||||
}
|
||||
|
||||
async run(input: WorkspaceGitScanInput): Promise<WorkspaceGitScanResult> {
|
||||
if (input.signal?.aborted) throw abortError("unresolved");
|
||||
let canonicalWorkspacePath: string;
|
||||
try {
|
||||
canonicalWorkspacePath = await fs.realpath(input.workspacePath);
|
||||
} catch (error) {
|
||||
throw new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.failed,
|
||||
"Workspace Git scan path is unavailable",
|
||||
{ cause: error instanceof Error ? error.message : String(error) },
|
||||
);
|
||||
}
|
||||
if (input.signal?.aborted) throw abortError(workspaceIdentity(canonicalWorkspacePath));
|
||||
|
||||
const workspaceHash = workspaceIdentity(canonicalWorkspacePath);
|
||||
const timeoutMs = clampInteger(input.timeoutMs, this.timeoutMs, 1, 120_000);
|
||||
const maxStdoutBytes = clampInteger(input.maxStdoutBytes, this.maxStdoutBytes, 1, 128 * 1024 * 1024);
|
||||
const maxStderrBytes = clampInteger(input.maxStderrBytes, this.maxStderrBytes, 1, 128 * 1024 * 1024);
|
||||
const key = scanKey({
|
||||
canonicalWorkspacePath,
|
||||
args: input.args,
|
||||
env: input.env,
|
||||
timeoutMs,
|
||||
maxStdoutBytes,
|
||||
maxStderrBytes,
|
||||
});
|
||||
const cacheTtlMs = clampInteger(input.cacheTtlMs, this.defaultCacheTtlMs, 0, 60_000);
|
||||
// A correctness-sensitive caller that explicitly disables caching must not
|
||||
// consume a result populated earlier by the file browser.
|
||||
const cached = cacheTtlMs > 0 ? this.readCache(key) : null;
|
||||
if (cached) {
|
||||
this.totals.cacheHits += 1;
|
||||
logger.debug({
|
||||
event: "workspace_git_scan",
|
||||
operation: input.operation,
|
||||
workspaceHash,
|
||||
cacheHit: true,
|
||||
singleFlightJoined: false,
|
||||
activeCount: this.activeCount,
|
||||
queuedCount: this.queue.length,
|
||||
cacheHitCount: this.totals.cacheHits,
|
||||
}, "workspace Git scan cache hit");
|
||||
return {
|
||||
stdout: cached.stdout,
|
||||
stderr: cached.stderr,
|
||||
canonicalWorkspacePath,
|
||||
workspaceHash,
|
||||
cacheHit: true,
|
||||
singleFlightJoined: false,
|
||||
};
|
||||
}
|
||||
|
||||
const existing = this.inFlight.get(key);
|
||||
if (existing) {
|
||||
existing.joinCount += 1;
|
||||
this.totals.singleFlightJoins += 1;
|
||||
return this.addWaiter(existing, input.signal, true);
|
||||
}
|
||||
|
||||
if (this.activeCount >= this.concurrency && this.queue.length >= this.queueCapacity) {
|
||||
this.totals.saturated += 1;
|
||||
this.warnRateLimited("saturated", {
|
||||
event: "workspace_git_scan",
|
||||
operation: input.operation,
|
||||
workspaceHash,
|
||||
outcome: "saturated",
|
||||
activeCount: this.activeCount,
|
||||
queuedCount: this.queue.length,
|
||||
saturationCount: this.totals.saturated,
|
||||
}, "workspace Git scan queue saturated");
|
||||
throw new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.saturated,
|
||||
"Changed files are temporarily unavailable because the Git scan queue is full",
|
||||
{
|
||||
workspaceHash,
|
||||
activeCount: this.activeCount,
|
||||
queuedCount: this.queue.length,
|
||||
retryAfterSeconds: 1,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
const fairnessKeys = Array.from(new Set([
|
||||
`repository:${workspaceHash}`,
|
||||
...(input.fairnessKeys ?? []).filter(Boolean),
|
||||
])).sort();
|
||||
const scan: PendingScan = {
|
||||
key,
|
||||
operation: input.operation,
|
||||
canonicalWorkspacePath,
|
||||
workspaceHash,
|
||||
args: [...input.args],
|
||||
fairnessKeys,
|
||||
env: input.env,
|
||||
timeoutMs,
|
||||
maxStdoutBytes,
|
||||
maxStderrBytes,
|
||||
cacheTtlMs,
|
||||
enqueuedAt: this.now(),
|
||||
state: "queued",
|
||||
controller: new AbortController(),
|
||||
waiters: new Map(),
|
||||
joinCount: 0,
|
||||
};
|
||||
this.inFlight.set(key, scan);
|
||||
this.queue.push(scan);
|
||||
const promise = this.addWaiter(scan, input.signal, false);
|
||||
this.drain();
|
||||
return promise;
|
||||
}
|
||||
|
||||
private addWaiter(
|
||||
scan: PendingScan,
|
||||
signal: AbortSignal | undefined,
|
||||
joined: boolean,
|
||||
): Promise<WorkspaceGitScanResult> {
|
||||
if (signal?.aborted) return Promise.reject(abortError(scan.workspaceHash));
|
||||
return new Promise((resolve, reject) => {
|
||||
const waiter: Waiter = {
|
||||
id: Symbol("workspace-git-waiter"),
|
||||
signal,
|
||||
resolve,
|
||||
reject,
|
||||
joined,
|
||||
};
|
||||
if (signal) {
|
||||
waiter.onAbort = () => {
|
||||
this.removeWaiter(scan, waiter);
|
||||
reject(abortError(scan.workspaceHash));
|
||||
};
|
||||
signal.addEventListener("abort", waiter.onAbort, { once: true });
|
||||
}
|
||||
scan.waiters.set(waiter.id, waiter);
|
||||
});
|
||||
}
|
||||
|
||||
private removeWaiter(scan: PendingScan, waiter: Waiter): void {
|
||||
if (!scan.waiters.delete(waiter.id)) return;
|
||||
if (waiter.signal && waiter.onAbort) {
|
||||
waiter.signal.removeEventListener("abort", waiter.onAbort);
|
||||
}
|
||||
if (scan.waiters.size > 0) return;
|
||||
|
||||
if (scan.state === "queued") {
|
||||
const index = this.queue.indexOf(scan);
|
||||
if (index >= 0) this.queue.splice(index, 1);
|
||||
this.inFlight.delete(scan.key);
|
||||
this.totals.cancelled += 1;
|
||||
logger.debug({
|
||||
event: "workspace_git_scan",
|
||||
operation: scan.operation,
|
||||
workspaceHash: scan.workspaceHash,
|
||||
outcome: "cancelled_before_start",
|
||||
activeCount: this.activeCount,
|
||||
queuedCount: this.queue.length,
|
||||
}, "queued workspace Git scan cancelled");
|
||||
this.drain();
|
||||
this.pruneFairnessState();
|
||||
return;
|
||||
}
|
||||
// Detach the doomed single-flight entry immediately. A new caller arriving
|
||||
// while the child is terminating may enqueue a fresh scan instead of
|
||||
// joining an already-aborted promise; the active slot remains occupied
|
||||
// until the child actually closes.
|
||||
if (this.inFlight.get(scan.key) === scan) this.inFlight.delete(scan.key);
|
||||
scan.controller.abort();
|
||||
}
|
||||
|
||||
private drain(): void {
|
||||
while (this.activeCount < this.concurrency && this.queue.length > 0) {
|
||||
const index = this.nextFairQueueIndex();
|
||||
const scan = this.queue.splice(index, 1)[0]!;
|
||||
if (scan.waiters.size === 0) {
|
||||
this.inFlight.delete(scan.key);
|
||||
continue;
|
||||
}
|
||||
this.start(scan);
|
||||
}
|
||||
}
|
||||
|
||||
private nextFairQueueIndex(): number {
|
||||
let selectedIndex = 0;
|
||||
let selectedScore = Number.POSITIVE_INFINITY;
|
||||
let selectedEnqueuedAt = Number.POSITIVE_INFINITY;
|
||||
for (let index = 0; index < this.queue.length; index += 1) {
|
||||
const scan = this.queue[index]!;
|
||||
// Average recency across every dimension so a common repository/company
|
||||
// key cannot mask a fresh actor or issue. Repeatedly changing only one
|
||||
// dimension therefore cannot jump ahead of a wholly unserved group.
|
||||
const score = scan.fairnessKeys.reduce(
|
||||
(total, key) => total + (this.lastServedByFairnessKey.get(key) ?? 0),
|
||||
0,
|
||||
) / scan.fairnessKeys.length;
|
||||
if (score < selectedScore || (score === selectedScore && scan.enqueuedAt < selectedEnqueuedAt)) {
|
||||
selectedIndex = index;
|
||||
selectedScore = score;
|
||||
selectedEnqueuedAt = scan.enqueuedAt;
|
||||
}
|
||||
}
|
||||
return selectedIndex;
|
||||
}
|
||||
|
||||
private start(scan: PendingScan): void {
|
||||
scan.state = "running";
|
||||
this.activeCount += 1;
|
||||
this.totals.started += 1;
|
||||
this.serviceSequence += 1;
|
||||
for (const key of scan.fairnessKeys) this.lastServedByFairnessKey.set(key, this.serviceSequence);
|
||||
const startedAt = this.now();
|
||||
const queueWaitMs = Math.max(0, startedAt - scan.enqueuedAt);
|
||||
|
||||
void this.runner({
|
||||
canonicalWorkspacePath: scan.canonicalWorkspacePath,
|
||||
args: scan.args,
|
||||
env: scan.env,
|
||||
signal: scan.controller.signal,
|
||||
timeoutMs: scan.timeoutMs,
|
||||
killGraceMs: this.killGraceMs,
|
||||
maxStdoutBytes: scan.maxStdoutBytes,
|
||||
maxStderrBytes: scan.maxStderrBytes,
|
||||
}).then(
|
||||
(result) => this.finishSuccess(scan, result, queueWaitMs, startedAt),
|
||||
(error) => this.finishFailure(scan, error, queueWaitMs, startedAt),
|
||||
);
|
||||
}
|
||||
|
||||
private finishSuccess(
|
||||
scan: PendingScan,
|
||||
result: WorkspaceGitRunnerResult,
|
||||
queueWaitMs: number,
|
||||
startedAt: number,
|
||||
): void {
|
||||
this.totals.succeeded += 1;
|
||||
if (scan.cacheTtlMs > 0) this.writeCache(scan.key, scan, result);
|
||||
const responseBase = {
|
||||
stdout: result.stdout,
|
||||
stderr: result.stderr,
|
||||
canonicalWorkspacePath: scan.canonicalWorkspacePath,
|
||||
workspaceHash: scan.workspaceHash,
|
||||
cacheHit: false,
|
||||
};
|
||||
for (const waiter of scan.waiters.values()) {
|
||||
this.detachWaiter(waiter);
|
||||
waiter.resolve({ ...responseBase, singleFlightJoined: waiter.joined });
|
||||
}
|
||||
logger.info({
|
||||
event: "workspace_git_scan",
|
||||
operation: scan.operation,
|
||||
workspaceHash: scan.workspaceHash,
|
||||
outcome: "success",
|
||||
queueWaitMs,
|
||||
executionMs: Math.max(0, this.now() - startedAt),
|
||||
activeCount: this.activeCount,
|
||||
queuedCount: this.queue.length,
|
||||
cacheHit: false,
|
||||
singleFlightJoinCount: scan.joinCount,
|
||||
exitOutcome: "zero",
|
||||
}, "workspace Git scan completed");
|
||||
this.release(scan);
|
||||
}
|
||||
|
||||
private finishFailure(
|
||||
scan: PendingScan,
|
||||
error: unknown,
|
||||
queueWaitMs: number,
|
||||
startedAt: number,
|
||||
): void {
|
||||
const normalized = isWorkspaceGitScanError(error)
|
||||
? error
|
||||
: new WorkspaceGitScanError(
|
||||
WORKSPACE_GIT_SCAN_ERROR_CODES.failed,
|
||||
"Workspace Git scan failed",
|
||||
{ cause: error instanceof Error ? error.message : String(error), workspaceHash: scan.workspaceHash },
|
||||
);
|
||||
if (normalized.code === WORKSPACE_GIT_SCAN_ERROR_CODES.timeout) {
|
||||
this.totals.timedOut += 1;
|
||||
this.warnRateLimited("timeout", {
|
||||
event: "workspace_git_scan",
|
||||
operation: scan.operation,
|
||||
workspaceHash: scan.workspaceHash,
|
||||
outcome: "timeout",
|
||||
queueWaitMs,
|
||||
executionMs: Math.max(0, this.now() - startedAt),
|
||||
activeCount: this.activeCount,
|
||||
queuedCount: this.queue.length,
|
||||
timeoutCount: this.totals.timedOut,
|
||||
}, "workspace Git scan timed out");
|
||||
} else if (normalized.code === WORKSPACE_GIT_SCAN_ERROR_CODES.cancelled) {
|
||||
this.totals.cancelled += 1;
|
||||
} else {
|
||||
this.totals.failed += 1;
|
||||
}
|
||||
for (const waiter of scan.waiters.values()) {
|
||||
this.detachWaiter(waiter);
|
||||
waiter.reject(normalized);
|
||||
}
|
||||
logger.info({
|
||||
event: "workspace_git_scan",
|
||||
operation: scan.operation,
|
||||
workspaceHash: scan.workspaceHash,
|
||||
outcome: normalized.code,
|
||||
queueWaitMs,
|
||||
executionMs: Math.max(0, this.now() - startedAt),
|
||||
activeCount: this.activeCount,
|
||||
queuedCount: this.queue.length,
|
||||
cacheHit: false,
|
||||
singleFlightJoinCount: scan.joinCount,
|
||||
exitOutcome: normalized.code,
|
||||
}, "workspace Git scan finished without a result");
|
||||
this.release(scan);
|
||||
}
|
||||
|
||||
private detachWaiter(waiter: Waiter): void {
|
||||
if (waiter.signal && waiter.onAbort) {
|
||||
waiter.signal.removeEventListener("abort", waiter.onAbort);
|
||||
}
|
||||
}
|
||||
|
||||
private release(scan: PendingScan): void {
|
||||
scan.waiters.clear();
|
||||
if (this.inFlight.get(scan.key) === scan) this.inFlight.delete(scan.key);
|
||||
this.activeCount = Math.max(0, this.activeCount - 1);
|
||||
this.drain();
|
||||
this.pruneFairnessState();
|
||||
}
|
||||
|
||||
private pruneFairnessState(): void {
|
||||
if (this.lastServedByFairnessKey.size === 0) return;
|
||||
const liveKeys = new Set<string>();
|
||||
for (const scan of this.inFlight.values()) {
|
||||
for (const key of scan.fairnessKeys) liveKeys.add(key);
|
||||
}
|
||||
for (const key of this.lastServedByFairnessKey.keys()) {
|
||||
if (!liveKeys.has(key)) this.lastServedByFairnessKey.delete(key);
|
||||
}
|
||||
}
|
||||
|
||||
private readCache(key: string): CacheEntry | null {
|
||||
const entry = this.cache.get(key);
|
||||
if (!entry) return null;
|
||||
if (entry.expiresAt <= this.now()) {
|
||||
this.deleteCacheEntry(key, entry);
|
||||
return null;
|
||||
}
|
||||
// LRU touch.
|
||||
this.cache.delete(key);
|
||||
this.cache.set(key, entry);
|
||||
return entry;
|
||||
}
|
||||
|
||||
private writeCache(key: string, scan: PendingScan, result: WorkspaceGitRunnerResult): void {
|
||||
if (this.maxCacheEntries === 0 || this.maxCacheBytes === 0) return;
|
||||
const bytes = Buffer.byteLength(result.stdout) + Buffer.byteLength(result.stderr);
|
||||
if (bytes > this.maxCacheBytes) return;
|
||||
const existing = this.cache.get(key);
|
||||
if (existing) this.deleteCacheEntry(key, existing);
|
||||
const entry: CacheEntry = {
|
||||
...result,
|
||||
canonicalWorkspacePath: scan.canonicalWorkspacePath,
|
||||
workspaceHash: scan.workspaceHash,
|
||||
expiresAt: this.now() + scan.cacheTtlMs,
|
||||
bytes,
|
||||
};
|
||||
this.cache.set(key, entry);
|
||||
this.cacheBytes += bytes;
|
||||
while (this.cache.size > this.maxCacheEntries || this.cacheBytes > this.maxCacheBytes) {
|
||||
const oldest = this.cache.entries().next().value as [string, CacheEntry] | undefined;
|
||||
if (!oldest) break;
|
||||
this.deleteCacheEntry(oldest[0], oldest[1]);
|
||||
}
|
||||
}
|
||||
|
||||
private deleteCacheEntry(key: string, entry: CacheEntry): void {
|
||||
if (!this.cache.delete(key)) return;
|
||||
this.cacheBytes = Math.max(0, this.cacheBytes - entry.bytes);
|
||||
}
|
||||
|
||||
private pruneExpiredCache(): void {
|
||||
const now = this.now();
|
||||
for (const [key, entry] of this.cache) {
|
||||
if (entry.expiresAt <= now) this.deleteCacheEntry(key, entry);
|
||||
}
|
||||
}
|
||||
|
||||
private warnRateLimited(
|
||||
key: string,
|
||||
fields: Record<string, unknown>,
|
||||
message: string,
|
||||
): void {
|
||||
const now = this.now();
|
||||
const bucket = this.warningBuckets.get(key);
|
||||
if (bucket && now - bucket.lastLoggedAt < this.warningIntervalMs) {
|
||||
bucket.suppressed += 1;
|
||||
return;
|
||||
}
|
||||
const suppressedSinceLastWarning = bucket?.suppressed ?? 0;
|
||||
this.warningBuckets.set(key, { lastLoggedAt: now, suppressed: 0 });
|
||||
logger.warn({ ...fields, suppressedSinceLastWarning }, message);
|
||||
}
|
||||
}
|
||||
|
||||
export function createWorkspaceGitOperationScheduler(
|
||||
options: WorkspaceGitOperationSchedulerOptions = {},
|
||||
): WorkspaceGitOperationScheduler {
|
||||
return new WorkspaceGitOperationScheduler(options);
|
||||
}
|
||||
|
||||
const envOptions = workspaceGitSchedulerOptionsFromEnv();
|
||||
export const workspaceGitOperationScheduler = createWorkspaceGitOperationScheduler(envOptions);
|
||||
|
||||
setExpensiveWorkspaceGitExecutor(async (input) => {
|
||||
const result = await workspaceGitOperationScheduler.run({
|
||||
workspacePath: input.localDir,
|
||||
args: input.args,
|
||||
operation: input.operation,
|
||||
cacheTtlMs: 0,
|
||||
timeoutMs: input.timeout,
|
||||
maxStdoutBytes: input.maxBuffer,
|
||||
maxStderrBytes: input.maxBuffer,
|
||||
});
|
||||
return { stdout: result.stdout, stderr: result.stderr };
|
||||
});
|
||||
|
||||
export function getWorkspaceGitOperationSchedulerSnapshot(): WorkspaceGitSchedulerSnapshot {
|
||||
return workspaceGitOperationScheduler.snapshot();
|
||||
}
|
||||
|
|
@ -53,6 +53,7 @@ import { workspaceOperationService, type WorkspaceOperationRecorder } from "./wo
|
|||
import { executionWorkspaceService, readExecutionWorkspaceConfig } from "./execution-workspaces.js";
|
||||
import { logActivity } from "./activity-log.js";
|
||||
import { readProjectWorkspaceRuntimeConfig } from "./project-workspace-runtime-config.js";
|
||||
import { workspaceGitOperationScheduler } from "./workspace-git-operation-scheduler.js";
|
||||
import {
|
||||
cleanupWorktreeInstanceArtifacts,
|
||||
deriveWorktreeInstanceId,
|
||||
|
|
@ -868,6 +869,22 @@ async function runGit(args: string[], cwd: string, opts?: { env?: NodeJS.Process
|
|||
return proc.stdout.trim();
|
||||
}
|
||||
|
||||
async function runExpensiveGitStatus(input: {
|
||||
args: readonly string[];
|
||||
cwd: string;
|
||||
operation: string;
|
||||
fairnessKeys?: readonly string[];
|
||||
}): Promise<string> {
|
||||
const result = await workspaceGitOperationScheduler.run({
|
||||
workspacePath: input.cwd,
|
||||
args: input.args,
|
||||
operation: input.operation,
|
||||
fairnessKeys: input.fairnessKeys,
|
||||
cacheTtlMs: 0,
|
||||
});
|
||||
return result.stdout.trim();
|
||||
}
|
||||
|
||||
function formatShortSha(value: string | null | undefined) {
|
||||
return value ? value.slice(0, 12) : "unknown";
|
||||
}
|
||||
|
|
@ -1294,10 +1311,15 @@ async function inspectGitWorktreeBranchIncoherence(input: {
|
|||
sourceIssue: ExecutionWorkspaceIssueRef | null;
|
||||
executionWorkspaceId?: string | null;
|
||||
}): Promise<GitWorktreeBranchIncoherenceEvidence> {
|
||||
const status = await runGit(
|
||||
["status", "--porcelain", "--untracked-files=all"],
|
||||
input.worktreePath,
|
||||
).catch(() => null);
|
||||
const status = await runExpensiveGitStatus({
|
||||
args: ["status", "--porcelain", "--untracked-files=all"],
|
||||
cwd: input.worktreePath,
|
||||
operation: "workspace_runtime.branch_incoherence_status",
|
||||
fairnessKeys: [
|
||||
...(input.executionWorkspaceId ? [`workspace:${input.executionWorkspaceId}`] : []),
|
||||
...(input.sourceIssue?.id ? [`issue:${input.sourceIssue.id}`] : []),
|
||||
],
|
||||
}).catch(() => null);
|
||||
const statusLines = status === null
|
||||
? null
|
||||
: status.split(/\r?\n/).map((line) => line.trimEnd()).filter((line) => line.trim().length > 0);
|
||||
|
|
@ -1781,7 +1803,15 @@ async function quarantineDirtyWorktreeBranchIncoherence(input: {
|
|||
`dirty quarantine repair checked out ${formatBranchForMessage(repairedBranch)} instead of ${input.expectedBranchName}`;
|
||||
throw branchIncoherenceValidationFailure(input.evidence);
|
||||
}
|
||||
const repairedStatus = await runGit(["status", "--porcelain", "--untracked-files=all"], input.worktreePath);
|
||||
const repairedStatus = await runExpensiveGitStatus({
|
||||
args: ["status", "--porcelain", "--untracked-files=all"],
|
||||
cwd: input.worktreePath,
|
||||
operation: "workspace_runtime.dirty_quarantine_verify",
|
||||
fairnessKeys: [
|
||||
...(input.executionWorkspaceId ? [`workspace:${input.executionWorkspaceId}`] : []),
|
||||
...(input.sourceIssue?.id ? [`issue:${input.sourceIssue.id}`] : []),
|
||||
],
|
||||
});
|
||||
if (repairedStatus.trim().length > 0) {
|
||||
input.evidence.safeRepair.succeeded = false;
|
||||
input.evidence.safeRepair.reason = "dirty quarantine repair completed but the worktree is still dirty";
|
||||
|
|
@ -2372,10 +2402,14 @@ async function refreshUnstartedWorktreeToBase(input: {
|
|||
// Force `--untracked-files=all` so untracked files are counted regardless of a
|
||||
// local `status.showUntrackedFiles=no`; otherwise the clean-tree guard could
|
||||
// pass and the `reset --hard` below would destroy untracked work.
|
||||
const status = await runGit(
|
||||
["status", "--porcelain", "--untracked-files=all"],
|
||||
input.worktreePath,
|
||||
).catch(() => null);
|
||||
const status = await runExpensiveGitStatus({
|
||||
args: ["status", "--porcelain", "--untracked-files=all"],
|
||||
cwd: input.worktreePath,
|
||||
operation: "workspace_runtime.base_refresh_clean_guard",
|
||||
fairnessKeys: [
|
||||
...(input.branchName ? [`branch:${input.branchName}`] : []),
|
||||
],
|
||||
}).catch(() => null);
|
||||
if (status === null || status.trim().length > 0) {
|
||||
return { refreshed: false, baseRefSha: null };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@ import type {
|
|||
WorkspaceFileListResponse,
|
||||
WorkspaceFileSelector,
|
||||
} from "@paperclipai/shared";
|
||||
import { api } from "./client";
|
||||
import { api, type RequestOptions } from "./client";
|
||||
|
||||
export interface FileResourceQuery {
|
||||
path: string;
|
||||
|
|
@ -50,11 +50,16 @@ export function buildFileResourceDownloadUrl(issueId: string, query: FileResourc
|
|||
}
|
||||
|
||||
export const fileResourcesApi = {
|
||||
list(issueId: string, query: FileResourceListQuery = {}): Promise<WorkspaceFileListResponse> {
|
||||
list(
|
||||
issueId: string,
|
||||
query: FileResourceListQuery = {},
|
||||
options?: RequestOptions,
|
||||
): Promise<WorkspaceFileListResponse> {
|
||||
const search = buildQuery(query);
|
||||
const suffix = search ? `?${search}` : "";
|
||||
return api.get<WorkspaceFileListResponse>(
|
||||
`/issues/${encodeURIComponent(issueId)}/file-resources/list${suffix}`,
|
||||
options,
|
||||
);
|
||||
},
|
||||
|
||||
|
|
|
|||
|
|
@ -864,6 +864,7 @@ export function FileViewerSheet({
|
|||
selectedPath={state.path}
|
||||
selectedProjectId={state.projectId}
|
||||
selectedWorkspaceId={state.workspaceId}
|
||||
active={computedOpen}
|
||||
className="min-h-0 flex-1 p-2"
|
||||
/>
|
||||
</aside>
|
||||
|
|
@ -914,6 +915,7 @@ export function FileViewerSheet({
|
|||
initialFolderPath={viewer.folderPath}
|
||||
initialProjectId={viewer.browseProjectId}
|
||||
initialWorkspaceId={viewer.browseWorkspaceId}
|
||||
active={computedOpen}
|
||||
className="min-h-0 flex-1 p-4"
|
||||
/>
|
||||
) : null}
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import type { Root } from "react-dom/client";
|
|||
import type { Project, ProjectWorkspace, WorkspaceFileListDirectoryItem, WorkspaceFileListFileItem, WorkspaceFileListItem, WorkspaceFileListResponse } from "@paperclipai/shared";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { WorkspaceFileBrowser, describeUnavailable } from "./WorkspaceFileBrowser";
|
||||
import { ApiError } from "@/api/client";
|
||||
|
||||
function act(callback: () => void | Promise<void>) {
|
||||
let result: void | Promise<void> | undefined;
|
||||
|
|
@ -34,6 +35,7 @@ async function waitForExpectation(assertion: () => void) {
|
|||
}
|
||||
|
||||
const useQueryMock = vi.fn();
|
||||
const cancelQueriesMock = vi.fn(() => Promise.resolve());
|
||||
const LIST_LIMIT = 100;
|
||||
|
||||
vi.mock("@tanstack/react-query", async () => {
|
||||
|
|
@ -42,6 +44,7 @@ vi.mock("@tanstack/react-query", async () => {
|
|||
...actual,
|
||||
useQuery: (options: unknown) => useQueryMock(options),
|
||||
useQueries: ({ queries }: { queries: unknown[] }) => queries.map((options) => useQueryMock(options)),
|
||||
useQueryClient: () => ({ cancelQueries: cancelQueriesMock }),
|
||||
};
|
||||
});
|
||||
|
||||
|
|
@ -205,6 +208,11 @@ describe("WorkspaceFileBrowser", () => {
|
|||
container = document.createElement("div");
|
||||
document.body.appendChild(container);
|
||||
useQueryMock.mockReset();
|
||||
cancelQueriesMock.mockClear();
|
||||
Object.defineProperty(document, "visibilityState", {
|
||||
configurable: true,
|
||||
value: "visible",
|
||||
});
|
||||
Object.defineProperty(Element.prototype, "scrollIntoView", {
|
||||
configurable: true,
|
||||
value: vi.fn(),
|
||||
|
|
@ -230,6 +238,107 @@ describe("WorkspaceFileBrowser", () => {
|
|||
return { root, onOpen };
|
||||
}
|
||||
|
||||
function changedListQueryOptions() {
|
||||
return useQueryMock.mock.calls
|
||||
.map(([options]) => options as {
|
||||
queryKey?: readonly unknown[];
|
||||
enabled?: boolean;
|
||||
refetchOnWindowFocus?: boolean;
|
||||
refetchOnReconnect?: boolean;
|
||||
})
|
||||
.filter((options) => options.queryKey?.[0] === "issues" && options.queryKey?.[3] === "list")
|
||||
.at(-1);
|
||||
}
|
||||
|
||||
it("does not enable changed-file enumeration before the Files panel opens", () => {
|
||||
useQueryMock.mockReturnValue(ok(availableResponse([])));
|
||||
|
||||
renderBrowser(vi.fn(), { active: false });
|
||||
|
||||
expect(changedListQueryOptions()).toMatchObject({
|
||||
enabled: false,
|
||||
refetchOnWindowFocus: false,
|
||||
refetchOnReconnect: false,
|
||||
});
|
||||
expect(cancelQueriesMock).toHaveBeenCalledWith({
|
||||
queryKey: ["issues", "file-resources", "issue-1", "list"],
|
||||
});
|
||||
});
|
||||
|
||||
it("disables and cancels changed-file enumeration while the page is hidden", () => {
|
||||
Object.defineProperty(document, "visibilityState", {
|
||||
configurable: true,
|
||||
value: "hidden",
|
||||
});
|
||||
useQueryMock.mockReturnValue(ok(availableResponse([])));
|
||||
|
||||
renderBrowser();
|
||||
|
||||
expect(changedListQueryOptions()?.enabled).toBe(false);
|
||||
expect(cancelQueriesMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("cancels an active changed-file query when the page becomes hidden", () => {
|
||||
useQueryMock.mockReturnValue(ok(availableResponse([])));
|
||||
renderBrowser();
|
||||
expect(changedListQueryOptions()?.enabled).toBe(true);
|
||||
|
||||
Object.defineProperty(document, "visibilityState", {
|
||||
configurable: true,
|
||||
value: "hidden",
|
||||
});
|
||||
act(() => {
|
||||
document.dispatchEvent(new Event("visibilitychange"));
|
||||
});
|
||||
|
||||
expect(changedListQueryOptions()?.enabled).toBe(false);
|
||||
expect(cancelQueriesMock).toHaveBeenCalledWith({
|
||||
queryKey: ["issues", "file-resources", "issue-1", "list"],
|
||||
});
|
||||
});
|
||||
|
||||
it("runs one explicit refresh without enabling focus or reconnect bursts", () => {
|
||||
const refetch = vi.fn();
|
||||
useQueryMock.mockReturnValue({
|
||||
...ok(availableResponse([createItem()])),
|
||||
refetch,
|
||||
});
|
||||
|
||||
renderBrowser();
|
||||
const refresh = container.querySelector<HTMLButtonElement>('button[aria-label="Refresh workspace files"]');
|
||||
expect(refresh).not.toBeNull();
|
||||
act(() => refresh!.click());
|
||||
|
||||
expect(refetch).toHaveBeenCalledTimes(1);
|
||||
expect(changedListQueryOptions()).toMatchObject({
|
||||
enabled: true,
|
||||
refetchOnWindowFocus: false,
|
||||
refetchOnReconnect: false,
|
||||
});
|
||||
});
|
||||
|
||||
it.each([
|
||||
[503, "workspace_git_scan_saturated"],
|
||||
[504, "workspace_git_scan_timeout"],
|
||||
] as const)("renders %s changed-file failures as a retryable state", (status, code) => {
|
||||
const refetch = vi.fn();
|
||||
useQueryMock.mockReturnValue({
|
||||
data: undefined,
|
||||
isFetching: false,
|
||||
isError: true,
|
||||
error: new ApiError("temporarily unavailable", status, { code }),
|
||||
refetch,
|
||||
});
|
||||
|
||||
renderBrowser();
|
||||
|
||||
expect(container.textContent).toContain("Changed files temporarily unavailable");
|
||||
const retry = Array.from(container.querySelectorAll("button")).find((button) => button.textContent?.includes("Retry"));
|
||||
expect(retry).not.toBeUndefined();
|
||||
act(() => retry!.click());
|
||||
expect(refetch).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("renders the Recently changed files as a tree and opens a row with its relative path", () => {
|
||||
useQueryMock.mockReturnValue(
|
||||
ok(availableResponse([createItem(), createItem({ relativePath: "README.md", displayPath: "README.md" })])),
|
||||
|
|
|
|||
|
|
@ -7,14 +7,16 @@ import {
|
|||
type KeyboardEvent,
|
||||
type ReactNode,
|
||||
} from "react";
|
||||
import { useQueries, useQuery } from "@tanstack/react-query";
|
||||
import { AlertTriangle, ChevronDown, ChevronRight, Cloud, Download, FileCode2, FolderOpen, Loader2, Search } from "lucide-react";
|
||||
import { useQueries, useQuery, useQueryClient } from "@tanstack/react-query";
|
||||
import { AlertTriangle, ChevronDown, ChevronRight, Cloud, Download, FileCode2, FolderOpen, Loader2, RefreshCcw, Search } from "lucide-react";
|
||||
import { Input } from "@/components/ui/input";
|
||||
import { Button } from "@/components/ui/button";
|
||||
import { cn } from "@/lib/utils";
|
||||
import { fileResourcesApi } from "@/api/file-resources";
|
||||
import { projectsApi } from "@/api/projects";
|
||||
import { ApiError } from "@/api/client";
|
||||
import { queryKeys } from "@/lib/queryKeys";
|
||||
import { usePageVisibility } from "@/lib/page-visibility";
|
||||
import { parseWorkspaceFileRef } from "@/lib/workspace-file-parser";
|
||||
import type {
|
||||
Project,
|
||||
|
|
@ -103,13 +105,24 @@ export function describeUnavailable(reason: string): { title: string; body: stri
|
|||
};
|
||||
}
|
||||
|
||||
function StateMessage({ icon, title, body }: { icon: ReactNode; title: string; body?: string }) {
|
||||
function StateMessage({
|
||||
icon,
|
||||
title,
|
||||
body,
|
||||
actions,
|
||||
}: {
|
||||
icon: ReactNode;
|
||||
title: string;
|
||||
body?: string;
|
||||
actions?: ReactNode;
|
||||
}) {
|
||||
return (
|
||||
<div className="flex items-start gap-3 px-1 py-8 text-sm">
|
||||
{icon}
|
||||
<div className="space-y-1">
|
||||
<p className="font-medium text-foreground">{title}</p>
|
||||
{body ? <p className="text-muted-foreground">{body}</p> : null}
|
||||
{actions ? <div className="pt-1">{actions}</div> : null}
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
|
|
@ -475,6 +488,8 @@ export interface WorkspaceFileBrowserProps {
|
|||
selectedPath?: string | null;
|
||||
selectedProjectId?: string | null;
|
||||
selectedWorkspaceId?: string | null;
|
||||
/** True only while the containing Files panel is actually open. */
|
||||
active?: boolean;
|
||||
className?: string;
|
||||
}
|
||||
|
||||
|
|
@ -491,8 +506,11 @@ export function WorkspaceFileBrowser({
|
|||
selectedPath,
|
||||
selectedProjectId: activeProjectId,
|
||||
selectedWorkspaceId: activeWorkspaceId,
|
||||
active = true,
|
||||
className,
|
||||
}: WorkspaceFileBrowserProps) {
|
||||
const queryClient = useQueryClient();
|
||||
const { visible: pageVisible } = usePageVisibility();
|
||||
const source: BrowserSource =
|
||||
initialProjectId && initialWorkspaceId ? "other" : "current";
|
||||
const workspace: WorkspaceFileSelector = "auto";
|
||||
|
|
@ -594,6 +612,7 @@ export function WorkspaceFileBrowser({
|
|||
const targetWorkspaceId = source === "other" ? selectedWorkspaceId : null;
|
||||
const effectiveWorkspace: WorkspaceFileSelector = source === "other" ? "project" : workspace;
|
||||
const canListFiles = source === "current" || Boolean(targetProjectId && targetWorkspaceId);
|
||||
const queriesEnabled = active && pageVisible && canListFiles;
|
||||
const targetRef = targetProjectId && targetWorkspaceId
|
||||
? { projectId: targetProjectId, workspaceId: targetWorkspaceId }
|
||||
: {};
|
||||
|
|
@ -607,8 +626,19 @@ export function WorkspaceFileBrowser({
|
|||
});
|
||||
}, [folderPath, onBrowseStateChange, searchInput, targetProjectId, targetWorkspaceId]);
|
||||
|
||||
const listQueryKey = useMemo(() => queryKeys.issues.fileResources(issueId, {
|
||||
workspace: effectiveWorkspace,
|
||||
projectId: targetProjectId,
|
||||
workspaceId: targetWorkspaceId,
|
||||
mode,
|
||||
q,
|
||||
limit: LIST_LIMIT,
|
||||
offset: 0,
|
||||
path: folderPath,
|
||||
}), [effectiveWorkspace, folderPath, issueId, mode, q, targetProjectId, targetWorkspaceId]);
|
||||
const listQuery = useQuery({
|
||||
queryKey: queryKeys.issues.fileResources(issueId, {
|
||||
queryKey: listQueryKey,
|
||||
queryFn: ({ signal }) => fileResourcesApi.list(issueId, {
|
||||
workspace: effectiveWorkspace,
|
||||
projectId: targetProjectId,
|
||||
workspaceId: targetWorkspaceId,
|
||||
|
|
@ -617,22 +647,21 @@ export function WorkspaceFileBrowser({
|
|||
limit: LIST_LIMIT,
|
||||
offset: 0,
|
||||
path: folderPath,
|
||||
}),
|
||||
queryFn: () => fileResourcesApi.list(issueId, {
|
||||
workspace: effectiveWorkspace,
|
||||
projectId: targetProjectId,
|
||||
workspaceId: targetWorkspaceId,
|
||||
mode,
|
||||
q,
|
||||
limit: LIST_LIMIT,
|
||||
offset: 0,
|
||||
path: folderPath,
|
||||
}),
|
||||
enabled: canListFiles,
|
||||
}, { signal }),
|
||||
enabled: queriesEnabled,
|
||||
retry: false,
|
||||
staleTime: 15_000,
|
||||
refetchOnWindowFocus: false,
|
||||
refetchOnReconnect: false,
|
||||
});
|
||||
|
||||
useEffect(() => {
|
||||
if (active && pageVisible) return;
|
||||
void queryClient.cancelQueries({
|
||||
queryKey: ["issues", "file-resources", issueId, "list"],
|
||||
});
|
||||
}, [active, issueId, pageVisible, queryClient]);
|
||||
|
||||
const data = listQuery.data;
|
||||
const items = useMemo(() => data?.items ?? [], [data]);
|
||||
const workspaceLabel = data?.workspace?.workspaceLabel ?? null;
|
||||
|
|
@ -676,7 +705,7 @@ export function WorkspaceFileBrowser({
|
|||
offset: spec.offset,
|
||||
path: spec.path || null,
|
||||
}),
|
||||
queryFn: () => fileResourcesApi.list(issueId, {
|
||||
queryFn: ({ signal }: { signal: AbortSignal }) => fileResourcesApi.list(issueId, {
|
||||
workspace: effectiveWorkspace,
|
||||
projectId: targetProjectId,
|
||||
workspaceId: targetWorkspaceId,
|
||||
|
|
@ -685,10 +714,12 @@ export function WorkspaceFileBrowser({
|
|||
limit: LIST_LIMIT,
|
||||
offset: spec.offset,
|
||||
path: spec.path || null,
|
||||
}),
|
||||
enabled: canListFiles && isLazyBrowse,
|
||||
}, { signal }),
|
||||
enabled: queriesEnabled && isLazyBrowse,
|
||||
retry: false,
|
||||
staleTime: 15_000,
|
||||
refetchOnWindowFocus: false,
|
||||
refetchOnReconnect: false,
|
||||
})),
|
||||
});
|
||||
const lazyItemsByFolder = useMemo(() => {
|
||||
|
|
@ -989,15 +1020,37 @@ export function WorkspaceFileBrowser({
|
|||
);
|
||||
} else if (listQuery.isError) {
|
||||
const status = listQuery.error instanceof ApiError ? listQuery.error.status : 0;
|
||||
const errorCode = listQuery.error instanceof ApiError && listQuery.error.body && typeof listQuery.error.body === "object"
|
||||
? (listQuery.error.body as { code?: unknown }).code
|
||||
: null;
|
||||
const changedFilesTemporarilyUnavailable = mode === "changed" && (
|
||||
status === 503 ||
|
||||
status === 504 ||
|
||||
errorCode === "workspace_git_scan_saturated" ||
|
||||
errorCode === "workspace_git_scan_timeout"
|
||||
);
|
||||
body = (
|
||||
<StateMessage
|
||||
icon={<AlertTriangle aria-hidden="true" className="h-5 w-5 text-amber-500" />}
|
||||
title="Couldn't load files"
|
||||
title={changedFilesTemporarilyUnavailable ? "Changed files temporarily unavailable" : "Couldn't load files"}
|
||||
body={
|
||||
status === 404
|
||||
changedFilesTemporarilyUnavailable
|
||||
? "Paperclip is limiting workspace scans to keep the server responsive. Try again in a moment."
|
||||
: status === 404
|
||||
? "Workspace browsing isn't available for this issue."
|
||||
: "Something went wrong loading workspace files."
|
||||
}
|
||||
actions={changedFilesTemporarilyUnavailable ? (
|
||||
<Button
|
||||
type="button"
|
||||
variant="outline"
|
||||
size="sm"
|
||||
disabled={!queriesEnabled}
|
||||
onClick={() => void listQuery.refetch()}
|
||||
>
|
||||
<RefreshCcw aria-hidden="true" className="mr-1 h-3 w-3" /> Retry
|
||||
</Button>
|
||||
) : null}
|
||||
/>
|
||||
);
|
||||
} else if (data?.state === "unavailable") {
|
||||
|
|
@ -1035,28 +1088,42 @@ export function WorkspaceFileBrowser({
|
|||
|
||||
return (
|
||||
<div className={cn("flex min-h-0 min-w-0 flex-col gap-2", className)}>
|
||||
<div className="relative min-w-0 max-w-full overflow-hidden">
|
||||
<Search
|
||||
aria-hidden="true"
|
||||
className="pointer-events-none absolute left-2.5 top-1/2 h-3.5 w-3.5 -translate-y-1/2 text-muted-foreground"
|
||||
/>
|
||||
<Input
|
||||
ref={inputRef}
|
||||
type="search"
|
||||
value={searchInput}
|
||||
onChange={(event) => setSearchInput(event.target.value)}
|
||||
onKeyDown={handleSearchKeyDown}
|
||||
placeholder="Search files by name or path…"
|
||||
aria-label="Search workspace files"
|
||||
role="combobox"
|
||||
aria-expanded={items.length > 0}
|
||||
aria-controls={items.length > 0 ? listboxId : undefined}
|
||||
aria-activedescendant={activeOptionId}
|
||||
autoFocus={autoFocusSearch}
|
||||
autoComplete="off"
|
||||
spellCheck={false}
|
||||
className="h-8 w-full max-w-full min-w-0 pl-8 font-mono text-xs"
|
||||
/>
|
||||
<div className="flex min-w-0 max-w-full items-center gap-1 overflow-hidden">
|
||||
<div className="relative min-w-0 flex-1">
|
||||
<Search
|
||||
aria-hidden="true"
|
||||
className="pointer-events-none absolute left-2.5 top-1/2 h-3.5 w-3.5 -translate-y-1/2 text-muted-foreground"
|
||||
/>
|
||||
<Input
|
||||
ref={inputRef}
|
||||
type="search"
|
||||
value={searchInput}
|
||||
onChange={(event) => setSearchInput(event.target.value)}
|
||||
onKeyDown={handleSearchKeyDown}
|
||||
placeholder="Search files by name or path…"
|
||||
aria-label="Search workspace files"
|
||||
role="combobox"
|
||||
aria-expanded={items.length > 0}
|
||||
aria-controls={items.length > 0 ? listboxId : undefined}
|
||||
aria-activedescendant={activeOptionId}
|
||||
autoFocus={autoFocusSearch}
|
||||
autoComplete="off"
|
||||
spellCheck={false}
|
||||
className="h-8 w-full max-w-full min-w-0 pl-8 font-mono text-xs"
|
||||
/>
|
||||
</div>
|
||||
<Button
|
||||
type="button"
|
||||
variant="ghost"
|
||||
size="icon-sm"
|
||||
onClick={() => void listQuery.refetch()}
|
||||
disabled={!queriesEnabled || listQuery.isFetching}
|
||||
aria-label="Refresh workspace files"
|
||||
title="Refresh workspace files"
|
||||
className="h-8 w-8 shrink-0"
|
||||
>
|
||||
<RefreshCcw aria-hidden="true" className={cn("h-3.5 w-3.5", listQuery.isFetching && "animate-spin")} />
|
||||
</Button>
|
||||
</div>
|
||||
|
||||
<WorkspaceFileBreadcrumbs
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@
|
|||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
getPageVisibility,
|
||||
getPageVisibilitySnapshot,
|
||||
getVisibilityHeaderValue,
|
||||
subscribePageVisibility,
|
||||
} from "./page-visibility";
|
||||
|
|
@ -40,6 +41,14 @@ describe("getPageVisibility", () => {
|
|||
setFocused(true);
|
||||
expect(getPageVisibility()).toEqual({ visible: true, focused: true });
|
||||
});
|
||||
|
||||
it("reconciles the snapshot before a hidden tab's first subscription", () => {
|
||||
setVisibility("hidden");
|
||||
expect(getPageVisibilitySnapshot()).toEqual({ visible: false, focused: false });
|
||||
|
||||
setVisibility("visible");
|
||||
expect(getPageVisibilitySnapshot().visible).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("getVisibilityHeaderValue", () => {
|
||||
|
|
|
|||
|
|
@ -89,6 +89,11 @@ export function subscribePageVisibility(listener: () => void): () => void {
|
|||
|
||||
/** Snapshot for `useSyncExternalStore` — returns a referentially-stable object between changes. */
|
||||
export function getPageVisibilitySnapshot(): PageVisibility {
|
||||
// Browser visibility can change before the first subscriber attaches (for
|
||||
// example when a restored tab mounts while already hidden). Reconcile on
|
||||
// every snapshot read so the initial render cannot briefly enable polling.
|
||||
const next = getPageVisibility();
|
||||
if (!sameState(next, cached)) cached = next;
|
||||
return cached;
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue