diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index ec628794216..f9b92a5fedd 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -33,7 +33,7 @@ Do not mark an item done from a timer removal or a green check alone. | 15 | Automatic database backups | Shared scheduling, actual backup deadlines, existing checkpoint contract | Pending | | 16 | Aggregate maintenance tick | Retire enabled and disabled branches after all constituent owners migrate | Pending | | 17 | Grok routine dispatch | Committed reservations and feature changes; prepared-only recovery | [#17490](https://github.com/paperclipai/paperclip/pull/17490) merged after exact-head 5/5 reviews and passing CI | -| 18 | Abandoned import transfer spools | Creation/completion events, existing retention deadline, startup recovery | Pending; explicitly retained from the remaining-blocker checklist | +| 18 | Abandoned import transfer spools | Creation/completion events, existing retention deadline, startup recovery | [#17835](https://github.com/paperclipai/paperclip/pull/17835), implemented; remote acceptance pending | ## Stack and verification contract @@ -3455,3 +3455,19 @@ awaiting provider cleanup inline. All 84 startup and 20 real-database environmen settlement cases pass, plus workspace typechecking and the full build. This changes only the last PR; earlier stack qualification and unresolved scope remain as recorded above. Fresh CI is required and nothing is merged. + +## Slice 59: import spool cleanup + +Branch `codex/sleep-59-import-spool-cleanup` follows slice 58 at +`929c86e32e00d219b329741dd2c89c40b1a5f826`. Replace the hourly abandoned-upload +scan with root/run-scoped filesystem hints, transaction-aware ledger changes, +and saved 24-hour deadlines. Keep failed/applying files quiet and preserve the +atomic cancellation claim. Database failures no longer count as missing rows +that permit orphan deletion. No new table or external-wake exemption. + +The owning three complete suites pass 50 tests with real PostgreSQL and files; +server typechecking passes. The initial outage regression reproduced deletion, +and fixture failures during lifecycle testing are recorded in +[the producer and lifecycle audit](2026-10-26-import-spool-cleanup-sleep.md). +Workspace and remote checks remain separate acceptance gates. This is one +remaining original duty, not a declaration that the inventory or stack is done. diff --git a/doc/plans/2026-10-26-import-spool-cleanup-sleep.md b/doc/plans/2026-10-26-import-spool-cleanup-sleep.md new file mode 100644 index 00000000000..85d3f014dac --- /dev/null +++ b/doc/plans/2026-10-26-import-spool-cleanup-sleep.md @@ -0,0 +1,263 @@ +# Import spool cleanup without an hourly scan + +This is slice 59 of the recurring server work migration, based on slice 58 +at `929c86e32e00d219b329741dd2c89c40b1a5f826`. The original inventory includes +abandoned company-transfer files among the remaining application maintenance +loops. This slice does not complete the inventory or certify other stack PRs. + +## Problem and ownership + +The application previously ran `sweepAbandonedImportTransferSpools` at startup +and every hour. It listed the import directory and queried every transfer with +files, including failed transfers retained for a future retry and applying +transfers owned by live work. These retained files caused SQL indefinitely. + +`company-import-transfers.ts` owns filesystem layout and cleanup policy. +`company-transfer-runs.ts` owns all supported production ledger writes, including +compare-and-set cancellation. `import-spool-cleanup-work.ts` connects these +owners to the existing delivery coordinator and scoped deadline helper. +`app.ts` registers the worker and removes its local filesystem subscription at +shutdown. The coordinator owns non-overlap, retries, drain, and joining cleanup. +There is no new table, filesystem watcher, timer, or provider abstraction. + +## Durable truth and deadlines + +| Existing state | Obligation and next wake | +| --- | --- | +| No directory | None; no ledger read or future scan | +| Pending/running upload with files | Saved `updatedAt + 24h + 1ms`; revalidate and cancel atomically before deletion | +| Failed transfer | Retain verified files for retry; no recurring work | +| Applying transfer | Retain files for its live apply owner; no recurring work | +| Completed/cancelled transfer with files | Delete immediately; failure keeps a bounded retry | +| Directory with no ledger row | Directory mtime plus 24h, rounded to the first millisecond strictly after the cutoff | +| Database or filesystem read error | Keep discovery unresolved and retry; absence is not inferred | + +The extra millisecond preserves the existing strict idle cutoff. A resumed +transfer updates its row, so the cancellation guard prevents deletion based on +a stale read. Cleanup uses cancelled, not failed: failed transfers remain +claimable for apply and cannot be used as a deletion claim. Failed files remain +retained indefinitely under the existing policy. Permission errors no longer +count as an empty directory. Only ENOENT means absence. Symlinks and non-UUID +names are not followed as import directories. + +Startup still separately recovers stranded applying rows once. The deadline +worker never performs that recovery, because doing so would interrupt live +applies. This migration preserves that startup operation's existing admission +behavior; it does not expand the worker's authority to fail live runs. + +## Producer and boundary audit + +All production writes to `company_transfer_runs` are in its service. A local +transaction helper registers the existing database-work intent before each +update. It composes with outer transactions; savepoint release does not publish +premature cleanup. The insert allocates its UUID before registering intent in +the existing idempotency transaction. Methods covered are: + +- `resumeOrCreate` when it inserts; returning an existing row makes no change. +- `recordManifest`, `start`, `attachCompany`, `notePartUploadActivity`, and `completePart` (activity deadlines). +- `claimApply`, `releaseApplyClaim`, `complete`, `fail`, and `cancel` (eligibility). +- `cancelIfIdle` (deletion claim). +- `recoverStrandedApplyingRuns` (one startup-wide invalidation). + +Ordinary hints identify one run. No directory means the worker returns before +reading the ledger, so export writes and declarations without parts do not +trigger global scans or cleanup SQL. The shared ledger now pays transaction +intent overhead on these writes; notifications do not make writes free. + +The filesystem owner emits process-local root/run hints after part writes, +part removal, and whole-spool removal, including partial failures. A failed +rename can leave a directory even if the subsequent ledger update never happens. +Startup scans existing directories to reconstruct that obligation. The cleanup +read selects only ID, company ID, status, and update time, not the manifest or +completed-part history. Pre-company imports and orphan directories share an +instance-level fairness group; this label is not an authorization identity. + +The supported writer model is the server's existing single-process import and +apply flow. Independent SQL writers, another server sharing these directories, +and external filesystem edits have no live notification contract. Restart +reconstruction still sees persisted files. This slice does not claim otherwise. + +## Lifecycle and verification + +The worker retains scoped invalidations arriving during discovery or cleanup. +Drain suppresses passive SQL and filesystem discovery without consuming pending +work. A first upload during drain still reconciles on resume. Shutdown removes +the filesystem listener and joins already-running cleanup through the delivery +owner. Failed removals retain the scoped retry floor even when another write +invalidates their run. Empty, failed, and applying states disarm the worker. + +A real PostgreSQL regression first reproduced an unsafe preexisting behavior: +a failed ledger read was swallowed as a missing row, and cleanup deleted the +old directory. The repaired test requires the error to propagate and the bytes +to remain. Its first setup attempt could not find a root-level tsx binary; +the repository's CLI-installed runner prepared the dedicated database instead. + +All 50 tests across three complete suites pass: import worker lifecycle, import +routes, and transfer ledger. This includes real files and PostgreSQL for strict +expiry, scoped first writes, retained history, orphan reconstruction, outer +commit and rollback, apply-versus-cleanup races, permission errors, partial file +writes, concurrent wakes, retry through drain, and shutdown joining. Existing +route tests cover upload/apply concurrency and successful apply with failed file +cleanup. No tests in these three suites were skipped. Server typechecking passes. + +The first lifecycle run passed 10 of 14 cases. Four fixture failures came from +fake time expiring PostgreSQL socket timers, a drain fixture disabling both +execution and transaction reconciliation, counting old filesystem spy calls, +and comparing advanced fake time against real filesystem mtimes. The fixture +now keeps database driver timers out of the simulated lifecycle, models the +production drain contract, and starts fresh observation windows. Product +assertions were retained; three additional lifecycle cases also pass. + +Workspace typechecking, full build, and remote PR checks are tracked separately +in the PR. Full local `test:run` remains unverified due to the previously recorded +embedded PostgreSQL capacity limitation. Dedicated `_test` databases were used +for these complete suites. No host semaphore cleanup was repeated. + +## Sleep boundary and remaining scope + +This proves worker quiet and reactivation, not database or host suspension. +The existing host-sleep proof still treats company transfer rows conservatively. +The Cloud alarm contract has not been extended to import-file expiry. Internal +`nextWakeAt` alone is not proof of a persisted external alarm. Do not remove that +sleep blocker on the strength of this worker migration. + +Other original inventory duties, master reconciliation, and existing stack +review/CI debts remain separate work. No stack PR is merged by this slice. + +## Review repairs + +The first review identified two concrete gaps, each reproduced before its fix: + +1. One unreadable child directory rejected the entire startup discovery pass, + leaving another completed, readable upload untouched. Discovery now retains + healthy entries and represents the failed child as unresolved work. The + existing per-entry retry and error path handles that child independently. + A failure to list the root still retains global discovery, because its child + inventory is unknown. A database outage likewise never proves absence. +2. Re-running the route suite against the external test database failed on its + fixed company primary key. Company and actor identities are now unique to the + fixture. Teardown deletes exactly its actors' transfer rows and its company, + then closes the database and removes temporary files. It also handles partial + setup without trying to remove an undefined path. + +All 51 owning tests now pass, including 18 worker cases. The full 24-case route +suite passed again against the same database after the combined run. Server +typechecking also passes. The database-drain fault fixture now holds the read +failure until explicitly released: a single transient read failure can recover +on the first per-entry check, so it did not represent an outage through drain. +The assertion that files survive a persistent outage remains unchanged. + +PR: [#17835](https://github.com/paperclipai/paperclip/pull/17835). +Remote acceptance and workspace checks remain tracked in that PR; these repairs +do not resolve other stack findings. + +## Upload activity ordering + +A later review found that the successful file-write hint can arrive before +`completePart` refreshes the ledger timestamp. An already-old running upload +could then lose its newly verified bytes to cleanup. A real route regression +held the part receipt until the actual cleanup claim ran: the upload returned +410 instead of 200 before the repair. + +The route now calls the ledger owner's `notePartUploadActivity` after validating +the part's length and hash, and before writing its bytes. This uses the same +transaction notification helper and the same pending/running state guard. It +refreshes activity without recording a receipt for bytes not yet written. The +idle cancellation compare-and-set then misses if upload activity wins. If +cancellation already won, the update cannot revive the run; the existing 410 +and stray-file cleanup path remains. Applying and failed states are not revived. +The extra activity write is explicit per verified upload; it is not a new table, +queue, or distributed lock. A failed filesystem write keeps a concrete grace +period before cleanup and still emits its filesystem hint. + +All 52 owning tests pass after this repair. The new route test requires HTTP 200, +exact file bytes, and the committed part receipt under the controlled race. +Earlier review repairs and repeatable fixture evidence remain valid. + +## Current master integration and spool acceptance (2026-10-29) + +Accepted source `87549234b125464dc7fa216227fc9bfa1d93f3d7` contains reviewed parent `fb252c8b0e52ce355e7d98db2e392e4831f99c61`. +The parent merged as `c8b7afee06c183ab922d097de1183d61c0bc4387`. A normal merge adopts the actual squash +after proving complete parent-tree identity. The complete accepted child tree +stays unchanged; only this audit follows in the publication. + +The composition proof preserves 1,237 parent-only paths, eight owned-only paths, +11,521 identical paths, and two shared automatic merge results: app registration +and delivery queue names. The sole manual conflict is in the documentation +ledger. It keeps the parent's landed Grok status and the owned import-spool PR +entry. There is no manual production resolution. Newly added narration and an +orphaned old startup comment fragment were removed without changing behavior. +Packages equal the parent byte for byte, including profile 80 and preserved +historical profile fixtures. Other native worker services are unchanged. + +All 293 cases across eight complete server files pass with zero skips: + +- 18 real spool worker lifecycle cases exercise filesystem and database work. +- 25 HTTP import-transfer cases cover uploaded bytes, receipts, and lifecycle. +- Nine transfer-ledger cases cover its write and recovery contract. +- Eight cloud capability-floor cases preserve disabled import behavior. +- 114 portability cases preserve actual import/export consumer contracts. +- 92 startup cases preserve lifecycle wiring and readiness. +- 26 coordinator cases and one scoped deadline case cover shared mechanisms. + +This includes quiet absent and retained spool states without SQL or directory +scans, strict saved deadlines, resume/apply races, nested commit/rollback, restart +reconstruction, failed writes, permission and database errors, root/symlink +boundaries, shutdown joining, and first/concurrent writes during drain. Storage +is isolated with a task-owned PAPERCLIP_HOME and temporary spool roots. Three +fresh PostgreSQL 17 databases use all 412 journal entries through tag/index +0413_tired_kronos. No source adaptation, assertion, or timeout changes were needed. +All eight test-file hashes remain exact. + +Database build, server production typecheck/build, module boundaries, migration +order, and current Copilot identity pass. The test command takes 12.28 seconds. +The parent's 4,192 selected native/resource cases belong to its recorded source; +they are not presented as rerun or added to this slice's 293 cases. No full local +repository, live provider, or deployed sleep acceptance is claimed. + +Earlier remote trusted-runner and Greptile service errors remain historical +evidence. Neither is treated as a fresh pass. Publication must obtain current +CI, security, and exact-head Greptile 5/5. The older Ponytail Ship verdict names +a different head and is not current review evidence. Actual findings must be +resolved; an older verdict cannot replace the current gate. + +Supported ledger writes use their transaction-aware notifications. Supported +filesystem mutations use the spool owner callbacks. Direct writes from another +process need their own ingress and can otherwise be discovered only at startup. +Internal cleanup deadlines do not prove an external Cloud wake. Existing sleep +blockers remain; no database suspension, external alarm coverage, or cost savings +is claimed. + +## CI concurrency test repair (2026-10-29) + +Current-head CI on `284294a9ab453406cfb6d4ff4bafc9c403dd1c80` passed 2,192 server-shard cases and failed +one tool-access audience race case. The test assumed that two 50 ms sleeps +queued member suspension ahead of an empty audience replacement. That ordering +was not established. If the explicit empty replacement wins while membership +is still active, its company-wide audience is the authorized result. + +The repair changes only that three-case test fixture. Dedicated backend PIDs +are observed, and each queued writer must be blocked by the exact first +transaction before its lock is released. Cleanup is proved to queue first. +All original inactive-audience, HTTP denial, and no-token assertions remain. +Finally releases transaction gates, joins pending work, and closes dedicated +clients if a readiness assertion fails. Production authorization is unchanged. + +A controlled 500 ms delay before cleanup issues its lock query reproduces all +three old timing failures. The old fixture permits empty replacement to finish +first; observed rows confirm a successful explicit replacement and an empty +audience after membership cleanup. With the same controlled delay, the repaired +fixture passes all three suspension/archive/removal cases. The artificial delay +and diagnostic logging are not committed. The final test keeps its original +15-second limit and uses bounded real PostgreSQL lock observations. + +The complete 472-case tool-access file passes on source `5d974e5eb28fc3e28b5e75e36fb18c61582348d5` in 20.86 +seconds, with zero skips. The eight earlier consumer files remain hash-identical +and their 293 passing cases remain source-scoped evidence. This yields 765 +distinct passing cases across nine complete files; controlled reruns are not +counted again. The first local full-file run passed 471 cases and timed out one +separate stdio-administration HTTP case at 15 seconds. A single fresh full-file +diagnostic run, with sanitized database wait-state capture, passes all 472. +That local timeout's cause was not established. It remains a failed command in +the record; no timeout or assertion changed and no first-run aggregate pass is +claimed. Fresh publication-head CI and reviews must pass again. diff --git a/server/src/__tests__/company-import-transfer-routes.test.ts b/server/src/__tests__/company-import-transfer-routes.test.ts index dae45dbd46c..9dcc6def729 100644 --- a/server/src/__tests__/company-import-transfer-routes.test.ts +++ b/server/src/__tests__/company-import-transfer-routes.test.ts @@ -1,15 +1,17 @@ -import { createHash } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { promises as fs } from "node:fs"; import os from "node:os"; import path from "node:path"; import express from "express"; import request from "supertest"; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { companies, createDb } from "@paperclipai/db"; +import { eq, inArray } from "drizzle-orm"; +import { companies, companyTransferRuns, createDb } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { companyTransferRunService } from "../services/company-transfer-runs.js"; import { sweepAbandonedImportTransferSpools } from "../services/company-import-transfers.js"; @@ -87,12 +89,14 @@ vi.mock("../services/company-import-transfers.js", async (importOriginal) => { }; }); -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_IMPORT_TRANSFER_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; -const companyId = "11111111-1111-4111-8111-111111111111"; +const companyId = randomUUID(); +const actorKeys = new Set(); const TEST_USER_HEADER = "x-test-user-id"; function boardActor(userId: string) { @@ -241,7 +245,9 @@ describeEmbeddedPostgres("company import transfer routes", () => { built.use((req, _res, next) => { const header = req.headers[TEST_USER_HEADER]; const userId = typeof header === "string" && header.length > 0 ? header : "board-user-a"; - (req as any).actor = boardActor(userId); + const scopedUserId = `${userId}-${companyId}`; + actorKeys.add(`board:${scopedUserId}`); + (req as any).actor = boardActor(scopedUserId); next(); }); built.use("/api/companies", companyRoutes(db, undefined, { importTransferSpoolRoot: spoolRoot })); @@ -250,18 +256,26 @@ describeEmbeddedPostgres("company import transfer routes", () => { } beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-import-transfer-routes-"); - db = createDb(tempDb.connectionString); + if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-import-transfer-routes-"); + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString); // The mocked import result points at this company; the ledger's // company_id foreign key needs the row to exist. - await db.insert(companies).values({ id: companyId, name: "Chunked Import Target" }); + await db.insert(companies).values({ id: companyId, name: "Chunked Import Target", issuePrefix: `IT${companyId.slice(0, 6)}` }); spoolRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-import-transfer-spool-")); app = await createApp(); }); afterAll(async () => { - await tempDb?.cleanup(); - await fs.rm(spoolRoot, { recursive: true, force: true }); + try { + if (db) { + if (actorKeys.size) await db.delete(companyTransferRuns).where(inArray(companyTransferRuns.actorKey, [...actorKeys])); + await db.delete(companies).where(eq(companies.id, companyId)); + } + } finally { + await db?.$client.end({ timeout: 1 }); + await tempDb?.cleanup(); + if (spoolRoot) await fs.rm(spoolRoot, { recursive: true, force: true }); + } }); beforeEach(() => { @@ -886,6 +900,27 @@ describeEmbeddedPostgres("company import transfer routes", () => { expect(malformed.status).toBe(404); }); + it("preserves a verified upload when cleanup runs before its part receipt commits", async () => { + const zip = buildFixtureZip(); + const { body, slices } = declareTransfer(zip, Math.ceil(zip.length / 2)); + const created = await request(app).post("/api/companies/import/transfers").send(body); + const transferId = created.body.transferId as string; + await db.update(companyTransferRuns).set({ updatedAt: new Date(Date.now() - 25 * 3600_000) }) + .where(eq(companyTransferRuns.id, transferId)); + const record = companyTransferRunService.completePart.bind(companyTransferRunService); + const interleave = vi.spyOn(companyTransferRunService, "completePart").mockImplementationOnce(async (...args) => { + + + await sweepAbandonedImportTransferSpools(db, spoolRoot); + return record(...args); + }); + try { + expect((await putPart(transferId, 0, slices[0]!)).status).toBe(200); + expect(await fs.readFile(path.join(spoolRoot, transferId, "part-0"))).toEqual(slices[0]); + expect((await companyTransferRunService.getRun(db, transferId))!.completedParts).toEqual(["part-0"]); + } finally { interleave.mockRestore(); } + }); + it("sweeps abandoned spools and cancels their open runs", async () => { const zip = buildFixtureZip(); const { body, slices } = declareTransfer(zip, Math.ceil(zip.length / 2)); @@ -930,7 +965,7 @@ describeEmbeddedPostgres("company import transfer routes", () => { // — as if a part upload bumped the row right after that read — while the // guarded UPDATE sees the fresh row in the database and must miss. const real = (await companyTransferRunService.getRun(db, transferId))!; - const staleRead = vi.spyOn(companyTransferRunService, "getRun").mockResolvedValue({ + const staleRead = vi.spyOn(companyTransferRunService, "getCleanupState").mockResolvedValue({ ...real, updatedAt: new Date(Date.now() - 25 * 60 * 60 * 1000), }); diff --git a/server/src/__tests__/company-transfer-runs.test.ts b/server/src/__tests__/company-transfer-runs.test.ts index 08cf4f023cc..86d0049c3da 100644 --- a/server/src/__tests__/company-transfer-runs.test.ts +++ b/server/src/__tests__/company-transfer-runs.test.ts @@ -1,3 +1,5 @@ +import { randomUUID } from "node:crypto"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { createDb } from "@paperclipai/db"; import { sha256HexOfBytes } from "@paperclipai/shared/portability-hash"; @@ -10,7 +12,8 @@ import { type CompanyTransferPartsIndex, } from "../services/company-transfer-runs.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TRANSFER_LEDGER_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; const pngBytes = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]); @@ -20,11 +23,12 @@ const pngBytes = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a] // packer. Distinct content per call: identical content intentionally derives // the same idempotency key, which would make separate tests resume each // other's runs. +const fixtureId = randomUUID(); let manifestSeq = 0; function demoManifest(): CompanyTransferPartsIndex & { idempotencyKey: string } { manifestSeq += 1; return { - idempotencyKey: sha256HexOfBytes(Buffer.from(`transfer-run-fixture-${manifestSeq}`)), + idempotencyKey: sha256HexOfBytes(Buffer.from(`transfer-run-fixture-${fixtureId}-${manifestSeq}`)), chunks: [{ name: "chunks/0000.json.gz" }, { name: "chunks/0001.json.gz" }], blobs: [{ name: `blobs/${sha256HexOfBytes(pngBytes)}` }], }; @@ -35,11 +39,12 @@ describeEmbeddedPostgres("companyTransferRunService", () => { let db!: ReturnType; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-transfer-runs-"); - db = createDb(tempDb.connectionString); + if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-transfer-runs-"); + db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString); }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/import-spool-cleanup-work.test.ts b/server/src/__tests__/import-spool-cleanup-work.test.ts new file mode 100644 index 00000000000..abcbe94d6e0 --- /dev/null +++ b/server/src/__tests__/import-spool-cleanup-work.test.ts @@ -0,0 +1,283 @@ +import { randomUUID } from "node:crypto"; +import { promises as fs } from "node:fs"; +import { setTimeout as delay } from "node:timers/promises"; +import { eq } from "drizzle-orm"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { companyTransferRuns, createDb, type Db } from "@paperclipai/db"; +import { companyTransferRunService } from "../services/company-transfer-runs.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { registerImportSpoolCleanupWork } from "../services/import-spool-cleanup-work.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; +import { IMPORT_TRANSFER_SPOOL_MAX_AGE_MS, settleImportTransferSpool, sweepAbandonedImportTransferSpools, writeImportTransferPart } from "../services/company-import-transfers.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +describe("import spool cleanup", () => { + let db: Db, root: string; + let coordinator: ReturnType | undefined; + let worker: ReturnType | undefined; + let admitted = true; + const errors: unknown[] = []; + let temporary: Awaited> | undefined; + beforeAll(async () => { + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_IMPORT_SPOOL_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("import-spool-"); + db = createDb(external ?? temporary!.connectionString, { idleTimeoutSeconds: 0, connectTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + }, 30_000); + beforeEach(async () => { + admitted = true; errors.length = 0; + root = await fs.mkdtemp(path.join(tmpdir(), "import-spool-")); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + worker?.unsubscribe(); await coordinator?.stop(); worker = undefined; coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); await fs.rm(root, { recursive: true, force: true }); + expect(errors).toEqual([]); + }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + async function transfer() { + return (await companyTransferRunService.resumeOrCreate(db, { direction: "import", actorKey: "test", idempotencyKey: randomUUID(), containerRef: {} })).run; + } + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, + canReconcile: () => true, onError: error => errors.push(error) }); + worker = registerImportSpoolCleanupWork(db, coordinator, root, () => admitted); + await worker.ready; + } + async function settle() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); await delay(5); + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); + } + async function quiet(ms = 6 * 3600_000) { + await settle(); + const before = db.$client.options.debug; + const queries: string[] = []; + const readDirs = vi.spyOn(fs, "readdir").mockClear(), stats = vi.spyOn(fs, "lstat").mockClear(); + db.$client.options.debug = (_id, query) => { queries.push(query); }; + try { + await vi.advanceTimersByTimeAsync(ms); await delay(10); + expect(queries).toEqual([]); expect(readDirs).not.toHaveBeenCalled(); expect(stats).not.toHaveBeenCalled(); + } finally { db.$client.options.debug = before; readDirs.mockRestore(); stats.mockRestore(); } + } + async function spool(runId: string) { await writeImportTransferPart(root, runId, 0, Buffer.from("saved parts")); } + async function exists(runId: string) { return fs.lstat(path.join(root, runId)).then(() => true, () => false); } + + it("stays quiet with no directory and reactivates on the first uploaded part", async () => { + await fs.rm(root, { recursive: true }); + await start(); await quiet(); + const run = await transfer(); await companyTransferRunService.start(db, run.id); await spool(run.id); await settle(); + const current = await companyTransferRunService.getRun(db, run.id); + expect(coordinator!.nextWakeAt()).toBe(current!.updatedAt.getTime() + IMPORT_TRANSFER_SPOOL_MAX_AGE_MS + 1); + await quiet(3600_000); expect(await exists(run.id)).toBe(true); + await vi.advanceTimersByTimeAsync(24 * 3600_000); await settle(); + expect(await exists(run.id)).toBe(false); + expect((await companyTransferRunService.getRun(db, run.id))!.status).toBe("cancelled"); + expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + }); + + it("keeps failed and applying uploads without recurring SQL or filesystem discovery", async () => { + const failed = await transfer(), applying = await transfer(); + await spool(failed.id); await spool(applying.id); + await companyTransferRunService.fail(db, failed.id, "retryable"); + expect(await companyTransferRunService.claimApply(db, applying.id)).toBe(true); + await start(); expect(coordinator!.nextWakeAt()).toBeNull(); await quiet(); + expect(await exists(failed.id)).toBe(true); expect(await exists(applying.id)).toBe(true); + + await companyTransferRunService.complete(db, applying.id); await settle(); + expect(await exists(applying.id)).toBe(false); expect(await exists(failed.id)).toBe(true); await quiet(); + }); + + it("reconstructs an orphan's exact mtime deadline after restart", async () => { + const id = randomUUID(); await spool(id); + const stamp = new Date(Date.now() - IMPORT_TRANSFER_SPOOL_MAX_AGE_MS + 20_000); + await fs.utimes(path.join(root, id), stamp, stamp); + await start(); expect(coordinator!.nextWakeAt()).toBe(stamp.getTime() + IMPORT_TRANSFER_SPOOL_MAX_AGE_MS + 1); + await vi.advanceTimersByTimeAsync(21_000); await settle(); + expect(await exists(id)).toBe(false); await quiet(); + }); + + it("updates the deadline after committed activity and preserves the strict idle boundary", async () => { + const run = await transfer(); await spool(run.id); await start(); + const before = coordinator!.nextWakeAt()!; + await vi.advanceTimersByTimeAsync(3600_000); + await companyTransferRunService.completePart(db, run.id, "part-0"); await settle(); + expect(coordinator!.nextWakeAt()).toBeGreaterThan(before); + const current = await companyTransferRunService.getRun(db, run.id); + const boundary = current!.updatedAt.getTime() + IMPORT_TRANSFER_SPOOL_MAX_AGE_MS; + expect(await settleImportTransferSpool(db, root, run.id, { now: new Date(boundary) })).toBe(false); + expect(await settleImportTransferSpool(db, root, run.id, { now: new Date(boundary + 1) })).toBe(true); + }); + + it("a resume or apply winning the cleanup claim keeps the spool", async () => { + const run = await transfer(); await spool(run.id); + const old = new Date(Date.now() - 25 * 3600_000); + await db.update(companyTransferRuns).set({ updatedAt: old }).where(eq(companyTransferRuns.id, run.id)); + const claim = companyTransferRunService.cancelIfIdle.bind(companyTransferRunService); + vi.spyOn(companyTransferRunService, "cancelIfIdle").mockImplementationOnce(async (...args) => { + expect(await companyTransferRunService.claimApply(db, run.id)).toBe(true); + return claim(...args); + }); + expect(await settleImportTransferSpool(db, root, run.id)).toBe(false); + expect(await exists(run.id)).toBe(true); + expect((await companyTransferRunService.getRun(db, run.id))!.status).toBe("applying"); + }); + + it.each(["commit", "rollback"] as const)("defers a nested terminal change until outer %s", async outcome => { + const run = await transfer(); await spool(run.id); await companyTransferRunService.fail(db, run.id, "retained"); + await start(); await quiet(); + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const changed = new Promise(resolve => { entered = resolve; }); + const transaction = db.transaction(async tx => { + await companyTransferRunService.cancel(tx as unknown as Db, run.id); entered(); await gate; + if (outcome === "rollback") throw new Error("rollback fixture"); + }).catch(error => { if (outcome !== "rollback" || error.message !== "rollback fixture") throw error; }); + await changed; + try { await vi.advanceTimersByTimeAsync(60_000); expect(await exists(run.id)).toBe(true); } + finally { release(); } + await transaction; await settle(); + expect(await exists(run.id)).toBe(outcome === "rollback"); await quiet(); + }); + + it("retains the first filesystem and ledger changes during drain without SQL", async () => { + await start(); await quiet(); admitted = false; + const run = await transfer(); await spool(run.id); await companyTransferRunService.cancel(db, run.id); + + await quiet(); expect(await exists(run.id)).toBe(true); + admitted = true; worker!.wake(); await settle(); + expect(await exists(run.id)).toBe(false); await quiet(); + }); + + it("retries failed cleanup, keeps its delay through a wake, and then becomes quiet", async () => { + const run = await transfer(); await spool(run.id); await companyTransferRunService.cancel(db, run.id); + const original = fs.rm.bind(fs); let denied = true; + const removal = vi.spyOn(fs, "rm").mockImplementation(async (target, options) => { + if (String(target) === path.join(root, run.id) && denied) throw Object.assign(new Error("denied"), { code: "EACCES" }); + return original(target, options); + }); + await start(); await settle(); expect(errors.splice(0)).toHaveLength(1); + expect(await exists(run.id)).toBe(true); + const attempts = removal.mock.calls.length; + await companyTransferRunService.complete(db, run.id); await settle(); + expect(removal).toHaveBeenCalledTimes(attempts); + denied = false; await vi.advanceTimersByTimeAsync(31_000); await settle(); + expect(await exists(run.id)).toBe(false); await quiet(); + }); + + it("does not call permission errors an empty directory", async () => { + vi.spyOn(fs, "readdir").mockRejectedValueOnce(Object.assign(new Error("directory denied"), { code: "EACCES" })); + await start(); expect(errors.splice(0)).toHaveLength(1); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await vi.advanceTimersByTimeAsync(31_000); await quiet(); + }); + + it("schedules cleanup after a failed part write leaves an orphan directory", async () => { + await start(); await quiet(); const id = randomUUID(); + vi.spyOn(fs, "rename").mockRejectedValueOnce(new Error("rename failed")); + await expect(spool(id)).rejects.toThrow("rename failed"); await settle(); + const stat = await fs.stat(path.join(root, id)); + expect(coordinator!.nextWakeAt()).toBe(Math.floor(stat.mtimeMs + IMPORT_TRANSFER_SPOOL_MAX_AGE_MS) + 1); + await vi.advanceTimersByTimeAsync(25 * 3600_000); await settle(); + expect(await exists(id)).toBe(false); await quiet(); + }); + + it("does not follow symlinks or schedule unrelated spool roots", async () => { + const other = await fs.mkdtemp(path.join(tmpdir(), "other-spool-")); + try { + const id = randomUUID(); await fs.symlink(other, path.join(root, id)); await start(); await quiet(); + await writeImportTransferPart(other, randomUUID(), 0, Buffer.from("other")); await quiet(); + expect(coordinator!.nextWakeAt()).toBeNull(); expect((await fs.lstat(path.join(root, id))).isSymbolicLink()).toBe(true); + } finally { await fs.rm(other, { recursive: true, force: true }); } + }); + + it("joins cleanup on shutdown and removes filesystem subscriptions", async () => { + const run = await transfer(); await spool(run.id); await companyTransferRunService.cancel(db, run.id); + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const began = new Promise(resolve => { entered = resolve; }); + const original = fs.rm.bind(fs); + vi.spyOn(fs, "rm").mockImplementation(async (target, options) => { + if (String(target) === path.join(root, run.id)) { entered(); await gate; } + return original(target, options); + }); + const starting = start(); await began; + worker!.unsubscribe(); let stopped = false; + const stopping = coordinator!.stop().then(() => { stopped = true; }); + await delay(10); expect(stopped).toBe(false); release(); await starting; await stopping; + expect(await exists(run.id)).toBe(false); + await spool(randomUUID()); expect(coordinator!.nextWakeAt()).toBeNull(); + }); + + it("keeps startup apply recovery separate from future live applies", async () => { + const stranded = await transfer(); await spool(stranded.id); + await companyTransferRunService.claimApply(db, stranded.id); + await companyTransferRunService.recoverStrandedApplyingRuns(db); + await start(); await quiet(); + expect((await companyTransferRunService.getRun(db, stranded.id))!.status).toBe("failed"); + expect(await companyTransferRunService.claimApply(db, stranded.id)).toBe(true); + await quiet(48 * 3600_000); + expect((await companyTransferRunService.getRun(db, stranded.id))!.status).toBe("applying"); + expect(await exists(stranded.id)).toBe(true); + }); + + it("retains another upload's wake while an earlier cleanup is running", async () => { + const first = await transfer(), second = await transfer(); + await spool(first.id); await companyTransferRunService.cancel(db, first.id); + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const began = new Promise(resolve => { entered = resolve; }); + const original = fs.rm.bind(fs); + vi.spyOn(fs, "rm").mockImplementation(async (target, options) => { + if (String(target) === path.join(root, first.id)) { entered(); await gate; } + return original(target, options); + }); + const starting = start(); await began; + try { await spool(second.id); await companyTransferRunService.cancel(db, second.id); } + finally { release(); } + await starting; await settle(); + expect(await exists(first.id)).toBe(false); expect(await exists(second.id)).toBe(false); await quiet(); + }); + + it("retains a failed database read through drain and recovers without losing the files", async () => { + const run = await transfer(); await spool(run.id); await companyTransferRunService.cancel(db, run.id); + const reader = vi.spyOn(companyTransferRunService, "getCleanupState").mockRejectedValue(new Error("database unavailable")); + await start(); expect(errors.splice(0)).toHaveLength(1); expect(await exists(run.id)).toBe(true); + admitted = false; await quiet(); expect(await exists(run.id)).toBe(true); + reader.mockRestore(); admitted = true; worker!.wake(); await settle(); + expect(await exists(run.id)).toBe(false); await quiet(); + }); + + it("cleans readable uploads while a different directory remains inaccessible", async () => { + const denied = await transfer(), healthy = await transfer(); + await spool(denied.id); await spool(healthy.id); + await companyTransferRunService.cancel(db, denied.id); await companyTransferRunService.cancel(db, healthy.id); + const original = fs.lstat.bind(fs); let inaccessible = true; + const stat = vi.spyOn(fs, "lstat").mockImplementation(async (...args) => { + if (String(args[0]) === path.join(root, denied.id) && inaccessible) + throw Object.assign(new Error("one directory denied"), { code: "EACCES" }); + return original(...args); + }); + await start(); await settle(); + expect(errors.splice(0)).toHaveLength(1); + expect(await exists(healthy.id)).toBe(false); + expect(await fs.readFile(path.join(root, denied.id, "part-0"), "utf8")).toBe("saved parts"); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + inaccessible = false; + await vi.advanceTimersByTimeAsync(31_000); await settle(); stat.mockRestore(); + expect(await exists(denied.id)).toBe(false); await quiet(); + }); + + it("does not treat a database outage as evidence that old files are orphaned", async () => { + const run = await transfer(); + await writeImportTransferPart(root, run.id, 0, Buffer.from("keep this upload")); + const old = new Date(Date.now() - 25 * 3600_000); + await fs.utimes(path.join(root, run.id), old, old); + vi.spyOn(db, "select").mockImplementation(() => { throw new Error("database unavailable"); }); + await expect(sweepAbandonedImportTransferSpools(db, root)).rejects.toThrow("database unavailable"); + expect(await fs.readFile(path.join(root, run.id, "part-0"), "utf8")).toBe("keep this upload"); + }); +}); diff --git a/server/src/__tests__/tool-access-service.test.ts b/server/src/__tests__/tool-access-service.test.ts index 37a00be75e6..00ba194591d 100644 --- a/server/src/__tests__/tool-access-service.test.ts +++ b/server/src/__tests__/tool-access-service.test.ts @@ -2061,6 +2061,13 @@ describeEmbeddedPostgres("tool access service", () => { maxConnections: 1, }); const replacementService = createTestToolAccessService(replacementDb); + const firstReplacementPid = (await firstReplacementDb.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`))[0]!.pid; + const cleanupPid = (await cleanupDb.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`))[0]!.pid; + const replacementPid = (await replacementDb.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`))[0]!.pid; + const waitForFirstReplacementLock = (waitingPid: number) => vi.waitFor(async () => { + const [waiting] = await db.execute<{ blockers: number[] }>(sql`select pg_blocking_pids(${waitingPid}) as blockers`); + expect(waiting!.blockers).toContain(firstReplacementPid); + }, { timeout: 5000 }); let releaseFirstReplacement!: () => void; const firstReplacementMayCommit = new Promise((resolve) => { releaseFirstReplacement = resolve; @@ -2116,70 +2123,76 @@ describeEmbeddedPostgres("tool access service", () => { } }); - // The first replacement still owns the departing member lock, so this - // gives cleanup time to queue for that lock before the empty replacement - // queues for the grant lock. - await new Promise((resolve) => setTimeout(resolve, 50)); - let replacementSettled = false; - const replacement = replacementService - .replaceConnectionGrantMembers(connection.id, grant.id, []) - .then( - (value) => ({ value, error: null }), - (error: unknown) => ({ value: null, error }), - ) - .finally(() => { - replacementSettled = true; - }); - await new Promise((resolve) => setTimeout(resolve, 50)); - expect(replacementSettled).toBe(false); + let replacementFinished: Promise | undefined; + try { + await waitForFirstReplacementLock(cleanupPid); + let replacementSettled = false; + const replacement = replacementService + .replaceConnectionGrantMembers(connection.id, grant.id, []) + .then( + (value) => ({ value, error: null }), + (error: unknown) => ({ value: null, error }), + ) + .finally(() => { + replacementSettled = true; + }); + replacementFinished = replacement; + await waitForFirstReplacementLock(replacementPid); + expect(replacementSettled).toBe(false); - releaseFirstReplacement(); - await firstReplacement; - await membershipIsLocked; - expect(replacementSettled).toBe(false); - releaseCleanup(); - await cleanup; - const replacementResult = await replacement; - expect(replacementResult.value).toBeNull(); - expect(replacementResult.error).toEqual( - expect.objectContaining({ - message: - "Replace inactive audience members before widening access to the whole company", - status: 409, - details: { - code: "audience_widening_blocked", - inactiveUserIds: ["departing-user"], - }, - }), - ); - expect( - await db - .select() - .from(connectionGrantMembers) - .where(eq(connectionGrantMembers.grantId, grant.id)), - ).toEqual([ - expect.objectContaining({ - subjectType: "user", - subjectId: "departing-user", - }), - ]); - - const fetchMock = vi - .spyOn(globalThis, "fetch") - .mockResolvedValue( - mcpHttpResponse({ token: "unexpected-company-wide-token" }), + releaseFirstReplacement(); + await firstReplacement; + await membershipIsLocked; + expect(replacementSettled).toBe(false); + releaseCleanup(); + await cleanup; + const replacementResult = await replacement; + expect(replacementResult.value).toBeNull(); + expect(replacementResult.error).toEqual( + expect.objectContaining({ + message: + "Replace inactive audience members before widening access to the whole company", + status: 409, + details: { + code: "audience_widening_blocked", + inactiveUserIds: ["departing-user"], + }, + }), ); - const denied = await request( - createRouteApp(db, agentJwtActor(company.id, agent.id, run.id)), - ) - .post(`/api/agents/me/connections/${connection.id}/token`) - .send({}); - expect(denied.status).toBe(403); - expect(denied.body).toMatchObject({ - code: "grant_audience_denied", - grantId: grant.id, - }); - expect(fetchMock).not.toHaveBeenCalled(); + expect( + await db + .select() + .from(connectionGrantMembers) + .where(eq(connectionGrantMembers.grantId, grant.id)), + ).toEqual([ + expect.objectContaining({ + subjectType: "user", + subjectId: "departing-user", + }), + ]); + + const fetchMock = vi + .spyOn(globalThis, "fetch") + .mockResolvedValue( + mcpHttpResponse({ token: "unexpected-company-wide-token" }), + ); + const denied = await request( + createRouteApp(db, agentJwtActor(company.id, agent.id, run.id)), + ) + .post(`/api/agents/me/connections/${connection.id}/token`) + .send({}); + expect(denied.status).toBe(403); + expect(denied.body).toMatchObject({ + code: "grant_audience_denied", + grantId: grant.id, + }); + expect(fetchMock).not.toHaveBeenCalled(); + } finally { + releaseFirstReplacement(); + releaseCleanup(); + await Promise.allSettled([firstReplacement, cleanup, ...(replacementFinished ? [replacementFinished] : [])]); + await Promise.all([firstReplacementDb.$client.end(), replacementDb.$client.end(), cleanupDb.$client.end()]); + } }, ); diff --git a/server/src/app.ts b/server/src/app.ts index a2ab7b062c3..dcb51306a6b 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -74,10 +74,9 @@ import { parseTrustProxyEnv, } from "./middleware/trust-proxy.js"; import { - IMPORT_TRANSFER_SPOOL_SWEEP_INTERVAL_MS, resolveDefaultImportTransferSpoolRoot, - sweepAbandonedImportTransferSpools, } from "./services/company-import-transfers.js"; +import { registerImportSpoolCleanupWork } from "./services/import-spool-cleanup-work.js"; import { companyTransferRunService } from "./services/company-transfer-runs.js"; import { healthRoutes } from "./routes/health.js"; import { livenessRoutes } from "./routes/liveness.js"; @@ -1233,38 +1232,11 @@ export async function createApp( ); emailChannels.start(); chatChannels.registerGitHubWebhookRecoveryWork(deliveryWork, () => !isWarmStandby() && !isIdleTaskDrainActive()); - // Abandoned chunked-import spool sweep: hourly (plus once at startup), - // deleting spool dirs whose transfer saw no activity for 24h and cancelling - // their still-open ledger runs. Same setInterval + unref + shutdown-clear - // shape as the feedback export flush above. - const importTransferSpoolRoot = resolveDefaultImportTransferSpoolRoot(); - const sweepImportTransferSpools = () => { - if (isWarmStandby() || isIdleTaskDrainActive()) return; - sweepAbandonedImportTransferSpools(db, importTransferSpoolRoot) - .then((result) => { - if (result.swept > 0) { - logger.info(result, "swept abandoned company import transfer spools"); - } - }) - .catch((err) => { - logger.error( - { err }, - "abandoned company import transfer spool sweep failed", - ); - }); - }; + const importSpoolCleanup = registerImportSpoolCleanupWork(db, deliveryWork, + resolveDefaultImportTransferSpoolRoot(), () => !isWarmStandby() && !isIdleTaskDrainActive()); registerBrowserUseCleanup(deliveryWork, browserUse, () => !isIdleTaskDrainActive()); - let importTransferSweepTimer: ReturnType | null = - setInterval( - sweepImportTransferSpools, - IMPORT_TRANSFER_SPOOL_SWEEP_INTERVAL_MS, - ); - importTransferSweepTimer.unref?.(); - // Startup only (never on the hourly interval — that would kill live - // applies): apply jobs are in-memory in this single process, so any run - // still "applying" now was interrupted by the previous shutdown and would - // otherwise 409 every retry forever. Fail those stranded runs — their - // spooled parts stay reusable — then run the normal sweep once. + + void companyTransferRunService .recoverStrandedApplyingRuns(db) .then((recovered) => { @@ -1277,9 +1249,6 @@ export async function createApp( }) .catch((err) => { logger.error({ err }, "stranded company transfer apply recovery failed"); - }) - .finally(() => { - sweepImportTransferSpools(); }); void toolDispatcher.initialize().catch((err) => { logger.error({ err }, "Failed to initialize plugin tool dispatcher"); @@ -1358,11 +1327,8 @@ export async function createApp( await dotMcpEvents?.stop(); jobCoordinator.stop(); resourceLifecycleDelivery.dispose(); + importSpoolCleanup.unsubscribe(); await deliveryWork.stop(); - if (importTransferSweepTimer) { - clearInterval(importTransferSweepTimer); - importTransferSweepTimer = null; - } devWatcher?.close(); viteHtmlRenderer?.dispose(); void viteDevServer?.close().catch(() => undefined); diff --git a/server/src/routes/companies.ts b/server/src/routes/companies.ts index 8de78282c20..768c003e787 100644 --- a/server/src/routes/companies.ts +++ b/server/src/routes/companies.ts @@ -886,6 +886,7 @@ export function companyRoutes(db: Db, storage?: StorageService, options?: Compan if (digest !== declared.sha256) { throw unprocessable(`Part ${partIndex} content does not match its declared sha256`); } + await companyTransferRunService.notePartUploadActivity(db, run.id); await writeImportTransferPart(importTransferSpoolRoot, run.id, partIndex, bytes); const recorded = await companyTransferRunService.completePart(db, run.id, partName); if (!recorded) { diff --git a/server/src/services/company-import-transfers.ts b/server/src/services/company-import-transfers.ts index 29b96045bac..0aba0433161 100644 --- a/server/src/services/company-import-transfers.ts +++ b/server/src/services/company-import-transfers.ts @@ -18,8 +18,6 @@ import { companyTransferRunService } from "./company-transfer-runs.js"; /** Spool dirs with no upload/apply activity for this long are abandoned. */ export const IMPORT_TRANSFER_SPOOL_MAX_AGE_MS = 24 * 60 * 60 * 1000; -/** How often the abandoned-spool sweep runs. */ -export const IMPORT_TRANSFER_SPOOL_SWEEP_INTERVAL_MS = 60 * 60 * 1000; const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; @@ -54,7 +52,19 @@ function partPathFor(spoolRoot: string, runId: string, index: number): string { return path.join(spoolDirFor(spoolRoot, runId), importTransferPartName(index)); } -/** Atomic part write: temp file in the same dir, then rename over the target. */ +const spoolListeners = new Set<(root: string, runId: string) => void>(); + +export function onImportTransferSpoolChange(listener: (root: string, runId: string) => void): () => void { + spoolListeners.add(listener); + return () => { spoolListeners.delete(listener); }; +} + +function changed(spoolRoot: string, runId: string) { + for (const listener of spoolListeners) listener(path.resolve(spoolRoot), runId); +} + + + export async function writeImportTransferPart( spoolRoot: string, runId: string, @@ -62,10 +72,15 @@ export async function writeImportTransferPart( bytes: Buffer, ): Promise { const target = partPathFor(spoolRoot, runId, index); - await fs.mkdir(path.dirname(target), { recursive: true }); - const tempPath = `${target}.tmp-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; - await fs.writeFile(tempPath, bytes); - await fs.rename(tempPath, target); + try { + await fs.mkdir(path.dirname(target), { recursive: true }); + const tempPath = `${target}.tmp-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`; + await fs.writeFile(tempPath, bytes); + await fs.rename(tempPath, target); + } finally { + + changed(spoolRoot, runId); + } } /** Byte size of a spooled part, or null when it is missing (or not a file). */ @@ -100,7 +115,9 @@ export async function assembleImportTransferZip( } export async function removeImportTransferSpool(spoolRoot: string, runId: string): Promise { - await fs.rm(spoolDirFor(spoolRoot, runId), { recursive: true, force: true }); + const dir = spoolDirFor(spoolRoot, runId); + try { await fs.rm(dir, { recursive: true, force: true }); } + finally { changed(spoolRoot, runId); } } /** @@ -112,72 +129,75 @@ export async function removeImportTransferPart( runId: string, index: number, ): Promise { - await fs.rm(partPathFor(spoolRoot, runId, index), { force: true }); + const target = partPathFor(spoolRoot, runId, index); + try { await fs.rm(target, { force: true }); } + finally { changed(spoolRoot, runId); } } -/** - * Delete spool dirs whose transfer saw no activity for `maxAgeMs` and cancel - * their still-open ledger runs. A run-backed spool is only deleted after an - * atomic claim: one guarded UPDATE cancels the run while it is still open - * (pending/running) AND its `updatedAt` (every part upload and resume bumps - * it) is still older than the cutoff — so a transfer that resumed after this - * sweep read the run makes the guard miss and keeps its spool. The claim - * cancels rather than fails on purpose: a failed run would still be claimable - * by `claimApply`, letting an apply assemble the spool while this sweep - * deletes it; a cancelled run is unclaimable the instant the UPDATE commits, - * before any file is removed. Terminal runs keep their status and their spool - * (a failed run's verified parts stay reusable for an apply retry; a - * completed run's spool was already removed at apply). Orphan dirs with no - * run row fall back to the dir's mtime — no run means no upload can race the - * delete. A swept run is never resumed: `resumeOrCreate` ignores cancelled - * runs, so redeclaring the same content starts a fresh run with every part - * missing — matching the 410 "re-create it" a part upload against the swept - * run gets. - */ -export async function sweepAbandonedImportTransferSpools( - db: Db, - spoolRoot: string, - options: { maxAgeMs?: number; now?: Date } = {}, -): Promise<{ swept: number }> { - const maxAgeMs = options.maxAgeMs ?? IMPORT_TRANSFER_SPOOL_MAX_AGE_MS; - const nowMs = (options.now ?? new Date()).getTime(); - const cutoff = new Date(nowMs - maxAgeMs); - let entries; + +export async function listImportTransferSpools(spoolRoot: string): Promise { try { - entries = await fs.readdir(spoolRoot, { withFileTypes: true }); - } catch { - return { swept: 0 }; + return (await fs.readdir(spoolRoot, { withFileTypes: true })) + .filter(entry => entry.isDirectory() && isImportTransferRunId(entry.name)).map(entry => entry.name); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return []; + throw error; } +} + +export async function readImportTransferSpool(db: Db, spoolRoot: string, runId: string) { + let stat; + try { stat = await fs.lstat(spoolDirFor(spoolRoot, runId)); } + catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + throw error; + } + if (!stat.isDirectory()) return null; + + const run = await companyTransferRunService.getCleanupState(db, runId); + return { run, mtimeMs: stat.mtimeMs }; +} + +type Spool = NonNullable>>; + +export function importTransferSpoolDeadline(spool: Spool, maxAgeMs = IMPORT_TRANSFER_SPOOL_MAX_AGE_MS): number | null { + if (!spool.run) return Math.floor(spool.mtimeMs + maxAgeMs) + 1; + if (spool.run.status === "completed" || spool.run.status === "cancelled") return 0; + if (spool.run.status === "pending" || spool.run.status === "running") + return spool.run.updatedAt.getTime() + maxAgeMs + 1; + + return null; +} + + + + +export async function settleImportTransferSpool( + db: Db, spoolRoot: string, runId: string, + options: { maxAgeMs?: number; now?: Date } = {}, +): Promise { + const spool = await readImportTransferSpool(db, spoolRoot, runId); + if (!spool) return false; + const maxAgeMs = options.maxAgeMs ?? IMPORT_TRANSFER_SPOOL_MAX_AGE_MS; + const now = options.now?.getTime() ?? Date.now(); + const deadline = importTransferSpoolDeadline(spool, maxAgeMs); + if (deadline === null || deadline > now) return false; + if (spool.run && spool.run.status !== "completed" && spool.run.status !== "cancelled") { + const claimed = await companyTransferRunService.cancelIfIdle(db, runId, new Date(now - maxAgeMs), + "Abandoned import transfer: spooled parts were deleted after prolonged inactivity."); + if (!claimed) return false; + } + await removeImportTransferSpool(spoolRoot, runId); + return true; +} + + +export async function sweepAbandonedImportTransferSpools( + db: Db, spoolRoot: string, options: { maxAgeMs?: number; now?: Date } = {}, +): Promise<{ swept: number }> { let swept = 0; - for (const entry of entries) { - if (!entry.isDirectory() || !isImportTransferRunId(entry.name)) continue; - const runId = entry.name; - const run = await companyTransferRunService.getRun(db, runId).catch(() => null); - if (run) { - if (run.status === "completed" || run.status === "cancelled") { - // Terminal runs never assemble again; a leftover spool (e.g. a - // best-effort post-apply cleanup that failed) is pure garbage and is - // collected immediately, regardless of idle time. - } else { - const claimed = await companyTransferRunService.cancelIfIdle( - db, - runId, - cutoff, - "Abandoned import transfer: spooled parts were deleted after prolonged inactivity.", - ); - if (!claimed) continue; - } - } else { - let mtimeMs: number; - try { - mtimeMs = (await fs.stat(path.join(spoolRoot, runId))).mtimeMs; - } catch { - continue; - } - if (nowMs - mtimeMs <= maxAgeMs) continue; - } - await fs.rm(path.join(spoolRoot, runId), { recursive: true, force: true }); - swept += 1; + for (const runId of await listImportTransferSpools(spoolRoot)) { + if (await settleImportTransferSpool(db, spoolRoot, runId, options)) swept++; } return { swept }; } diff --git a/server/src/services/company-transfer-runs.ts b/server/src/services/company-transfer-runs.ts index 8090da4990e..677ac7c9a0d 100644 --- a/server/src/services/company-transfer-runs.ts +++ b/server/src/services/company-transfer-runs.ts @@ -1,6 +1,8 @@ +import { randomUUID } from "node:crypto"; import { and, desc, eq, inArray, lt, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { companyTransferRuns } from "@paperclipai/db"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; // Durable run ledger for chunked company transfers. A run tracks one export // publish or import apply of a transfer container. Progress is recorded per @@ -128,9 +130,26 @@ async function getRun(db: Db, runId: string): Promise(db: Db, runId: string, write: (tx: Db) => Promise): Promise { + return db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.importSpoolCleanup, runId); + return write(tx as unknown as Db); + }); +} + export const companyTransferRunService = { getRun, + async getCleanupState(db: Db, runId: string) { + const [run] = await db.select({ id: companyTransferRuns.id, companyId: companyTransferRuns.companyId, + status: companyTransferRuns.status, updatedAt: companyTransferRuns.updatedAt }) + .from(companyTransferRuns).where(eq(companyTransferRuns.id, runId)).limit(1); + return run ?? null; + }, + async getRunForActor(db: Db, runId: string, actorKey: string): Promise { const run = await getRun(db, runId); return run && run.actorKey === actorKey ? run : null; @@ -171,9 +190,12 @@ export const companyTransferRunService = { if (existing) { return { run: existing, resumed: true, alreadyCompleted: existing.status === "completed" }; } + const id = randomUUID(); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.importSpoolCleanup, id); const [created] = await txDb .insert(companyTransferRuns) .values({ + id, direction: input.direction, actorKey: input.actorKey, idempotencyKey: input.idempotencyKey, @@ -200,7 +222,7 @@ export const companyTransferRunService = { const index = asPartsIndex(manifest); const chunkCount = counts?.chunkCount ?? index?.chunks.length ?? 0; const blobCount = counts?.blobCount ?? index?.blobs.length ?? 0; - await db + await writeTransferRun(db, runId, async tx => tx .update(companyTransferRuns) .set({ manifest, @@ -209,7 +231,7 @@ export const companyTransferRunService = { blobCount, updatedAt: new Date(), }) - .where(eq(companyTransferRuns.id, runId)); + .where(eq(companyTransferRuns.id, runId))); }, /** @@ -222,30 +244,39 @@ export const companyTransferRunService = { * not run the apply. */ async claimApply(db: Db, runId: string): Promise { - const claimed = await db + const claimed = await writeTransferRun(db, runId, async tx => tx .update(companyTransferRuns) .set({ status: "applying", error: null, updatedAt: new Date() }) .where( and(eq(companyTransferRuns.id, runId), inArray(companyTransferRuns.status, RESUMABLE_STATUSES)), ) - .returning({ id: companyTransferRuns.id }); + .returning({ id: companyTransferRuns.id })); return claimed.length > 0; }, async start(db: Db, runId: string): Promise { - await db + await writeTransferRun(db, runId, async tx => tx .update(companyTransferRuns) .set({ status: "running", error: null, startedAt: new Date(), finishedAt: null, updatedAt: new Date() }) .where( and(eq(companyTransferRuns.id, runId), inArray(companyTransferRuns.status, RESUMABLE_STATUSES)), - ); + )); }, async attachCompany(db: Db, runId: string, companyId: string): Promise { - await db + await writeTransferRun(db, runId, async tx => tx .update(companyTransferRuns) .set({ companyId, updatedAt: new Date() }) - .where(eq(companyTransferRuns.id, runId)); + .where(eq(companyTransferRuns.id, runId))); + }, + + + + + async notePartUploadActivity(db: Db, runId: string): Promise { + await writeTransferRun(db, runId, async tx => tx.update(companyTransferRuns) + .set({ updatedAt: new Date() }) + .where(and(eq(companyTransferRuns.id, runId), inArray(companyTransferRuns.status, PART_UPLOAD_STATUSES)))); }, /** @@ -258,7 +289,7 @@ export const companyTransferRunService = { */ async completePart(db: Db, runId: string, partName: string): Promise { const partJson = JSON.stringify([partName]); - const updated = await db + const updated = await writeTransferRun(db, runId, async tx => tx .update(companyTransferRuns) .set({ completedParts: sql`case when ${companyTransferRuns.completedParts} @> ${partJson}::jsonb then ${companyTransferRuns.completedParts} else ${companyTransferRuns.completedParts} || ${partJson}::jsonb end`, @@ -267,7 +298,7 @@ export const companyTransferRunService = { .where( and(eq(companyTransferRuns.id, runId), inArray(companyTransferRuns.status, PART_UPLOAD_STATUSES)), ) - .returning({ id: companyTransferRuns.id }); + .returning({ id: companyTransferRuns.id })); return updated.length > 0; }, @@ -285,10 +316,10 @@ export const companyTransferRunService = { }, async complete(db: Db, runId: string): Promise { - await db + await writeTransferRun(db, runId, async tx => tx .update(companyTransferRuns) .set({ status: "completed", error: null, finishedAt: new Date(), updatedAt: new Date() }) - .where(eq(companyTransferRuns.id, runId)); + .where(eq(companyTransferRuns.id, runId))); }, /** @@ -306,7 +337,7 @@ export const companyTransferRunService = { * same content afterwards starts a fresh run. */ async cancelIfIdle(db: Db, runId: string, idleSince: Date, error: string): Promise { - const claimed = await db + const claimed = await writeTransferRun(db, runId, async tx => tx .update(companyTransferRuns) .set({ status: "cancelled", error, finishedAt: new Date(), updatedAt: new Date() }) .where( @@ -316,7 +347,7 @@ export const companyTransferRunService = { lt(companyTransferRuns.updatedAt, idleSince), ), ) - .returning({ id: companyTransferRuns.id }); + .returning({ id: companyTransferRuns.id })); return claimed.length > 0; }, @@ -329,11 +360,11 @@ export const companyTransferRunService = { * run already settled this is a no-op and returns false. */ async releaseApplyClaim(db: Db, runId: string, error: string): Promise { - const released = await db + const released = await writeTransferRun(db, runId, async tx => tx .update(companyTransferRuns) .set({ status: "failed", error, finishedAt: new Date(), updatedAt: new Date() }) .where(and(eq(companyTransferRuns.id, runId), eq(companyTransferRuns.status, "applying"))) - .returning({ id: companyTransferRuns.id }); + .returning({ id: companyTransferRuns.id })); return released.length > 0; }, @@ -346,7 +377,7 @@ export const companyTransferRunService = { * applies may be live (e.g. from a periodic sweep) would kill them. */ async recoverStrandedApplyingRuns(db: Db): Promise { - const recovered = await db + const recovered = await writeTransferRun(db, "*", async tx => tx .update(companyTransferRuns) .set({ status: "failed", @@ -355,21 +386,21 @@ export const companyTransferRunService = { updatedAt: new Date(), }) .where(eq(companyTransferRuns.status, "applying")) - .returning({ id: companyTransferRuns.id }); + .returning({ id: companyTransferRuns.id })); return recovered.map((row) => row.id); }, async fail(db: Db, runId: string, error: string): Promise { - await db + await writeTransferRun(db, runId, async tx => tx .update(companyTransferRuns) .set({ status: "failed", error, finishedAt: new Date(), updatedAt: new Date() }) - .where(eq(companyTransferRuns.id, runId)); + .where(eq(companyTransferRuns.id, runId))); }, async cancel(db: Db, runId: string): Promise { - await db + await writeTransferRun(db, runId, async tx => tx .update(companyTransferRuns) .set({ status: "cancelled", finishedAt: new Date(), updatedAt: new Date() }) - .where(and(eq(companyTransferRuns.id, runId), inArray(companyTransferRuns.status, RESUMABLE_STATUSES))); + .where(and(eq(companyTransferRuns.id, runId), inArray(companyTransferRuns.status, RESUMABLE_STATUSES)))); }, }; diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 1b450c94681..f087fccba55 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + importSpoolCleanup: "import-spool-cleanup", externalConversationSettlement: "external-conversation-settlement", nativeRecoveryNotices: "native-recovery-notices", nativeEnvironmentSettlement: "native-environment-settlement", diff --git a/server/src/services/import-spool-cleanup-work.ts b/server/src/services/import-spool-cleanup-work.ts new file mode 100644 index 00000000000..082e114a95b --- /dev/null +++ b/server/src/services/import-spool-cleanup-work.ts @@ -0,0 +1,56 @@ +import path from "node:path"; +import type { Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; +import { importTransferSpoolDeadline, isImportTransferRunId, listImportTransferSpools, + onImportTransferSpoolChange, readImportTransferSpool, settleImportTransferSpool } from "./company-import-transfers.js"; + +type Entry = { id: string; companyId: string; dueAt: number }; + +export function registerImportSpoolCleanupWork(db: Db, coordinator: ReturnType, + spoolRoot: string, canRun: () => boolean) { + const root = path.resolve(spoolRoot); + async function read(id: string): Promise { + const spool = await readImportTransferSpool(db, root, id); + if (!spool) return null; + const dueAt = importTransferSpoolDeadline(spool); + + + return dueAt === null ? null : { id, companyId: spool.run?.companyId ?? "import-spools-without-company", dueAt }; + } + const worker = registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.importSpoolCleanup, + decodeScope: (hint: string) => { + if (hint === "*") return undefined; + if (!isImportTransferRunId(hint)) throw new Error("Invalid import spool scope"); + return hint; + }, + matches: (entry: Entry, scope?: string) => !scope || entry.id === scope, + load: async (scope, signal) => { + const entries: Entry[] = []; + for (const id of scope ? [scope] : await listImportTransferSpools(root)) { + if (signal?.aborted || !canRun()) throw new Error("Import spool discovery paused"); + try { + const entry = await read(id); + if (entry) entries.push(entry); + } catch { + + + + entries.push({ id, companyId: "import-spools-without-company", dueAt: Date.now() }); + } + } + return entries; + }, + check: async (entry, signal) => { + if (signal?.aborted || !canRun()) throw new Error("Import spool cleanup paused"); + await settleImportTransferSpool(db, root, entry.id); + return read(entry.id); + }, + }, canRun); + const unsubscribe = onImportTransferSpoolChange((changedRoot, runId) => { + if (changedRoot === root) worker.invalidate(runId); + }); + return { ...worker, unsubscribe }; +}