From 1bceae0c91a914d2047e03d3ad1b9c8cd9f71f12 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 29 Oct 2026 07:42:14 -0500 Subject: [PATCH] Make import spool cleanup wake on changes and saved deadlines (#17835) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit https://entire.com/gh/paperclipai/-/trails/38/changes/paperclip/38 ## Thinking Path > - Paperclip manages agent work and the files that support it. > - Resumable company imports keep uploaded parts until apply or expiry. > - An hourly cleanup loop reads the database for every retained upload directory. > - Failed and applying imports can retain files without owing periodic cleanup. > - Existing transfer rows and file timestamps already identify real deadlines. > - This change schedules those deadlines and wakes cleanup after relevant changes. ## Linked Issues or Issue Description Refs: #17801. **What problem does this solve?** The hourly import sweep keeps querying retained uploads when no cleanup is due. It also treats a failed database read as a missing transfer row. That can cause deletion of a live upload whose directory is old. **What is the proposed solution?** Use the existing delivery coordinator with per-transfer deadlines. Keep commit-aware notifications in the transfer ledger service. Keep filesystem change notifications in the spool owner. Recover pending cleanup from existing files and rows at startup. **What alternatives have you considered?** A slower periodic sweep would still query idle history. A new job table would duplicate the existing ledger and directory timestamps. Retrying only after successful uploads would miss partial file writes that leave cleanup work. This is slice 59. Parent #17801 has merged. This publication targets master and preserves that landed prerequisite. Fresh current-head CI and review remain required. ## What Changed - Repair an inherited CI concurrency test. Observe exact PostgreSQL blockers instead of assuming lock order from two 50 ms sleeps. Keep production authorization and all fail-closed assertions unchanged. - Refresh verified-upload activity before writing bytes. Preserve the guarded cleanup claim without recording a part receipt before the file exists. - Isolate unreadable child directories so healthy cleanup continues. Use unique, self-cleaning route fixtures for external database reruns. - Replace the hourly timer with the existing scoped deadline worker. No new table or independent scheduler. - Schedule open uploads at saved activity time plus 24 hours, preserving the strict cutoff and guarded cancellation before deletion. - Keep failed and applying imports quiet while retaining their files. Clean completed/cancelled leftovers immediately. - Register run-scoped write intent before every supported ledger mutation, including nested transactions. The helper stays inside the existing ledger owner. - Wake from filesystem changes, including failed writes and removals. Reconstruct orphan deadlines from directory timestamps after restart. - Propagate database and permission errors. Only a missing directory proves absence. Read only the four ledger fields needed for cleanup. - Preserve startup-only stranded-apply recovery. The cleanup worker cannot fail live applies. Remove its filesystem subscription and join active cleanup on shutdown. - Record producer coverage and limits in `doc/plans/2026-10-26-import-spool-cleanup-sleep.md`. ## Verification Current repair source: `5d974e5eb28fc3e28b5e75e36fb18c61582348d5`. The publication adds a test-fixture repair and audit text to the previously tested import-spool source; runtime code is unchanged. **765 distinct cases pass across nine complete files:** the original 293 consumer cases remain hash-identical, and the full tool-access file passes all 472 in 20.86 seconds. Controlled evidence: a 500 ms cleanup-query delay fails all three old timing cases; observed lock barriers pass all three with the same delay and unchanged policy assertions. The artificial delay is not committed. Current-head CI's original audience-race failure is preserved. The first local full-file run also retains one separate 15-second stdio HTTP timeout; one fresh diagnostic full run passes all 472, without establishing that timeout's cause or changing its limit. Fresh CI and reviews are required on this new publication head. Accepted production source: `87549234b125464dc7fa216227fc9bfa1d93f3d7`. Publication head: `016b3af51a734ad0ddf574de701679a7bdc33d06`. Adopting the actual parent squash preserves the complete accepted child tree. The subsequent changes are the CI test fixture and its audit text. - **293 cases pass across eight complete server files, with zero skips:** spool worker 18; HTTP import transfers 25; transfer ledger 9; cloud floor 8; portability 114; startup 92; coordinator 26; scoped deadlines 1. Runtime: 12.28 seconds. - Real filesystem and PostgreSQL cases cover zero idle SQL/directory scans, actual deadlines, first/concurrent writes, nested commit/rollback, restart, failure, drain/resume, root and symlink boundaries, and joined shutdown. Three fresh PostgreSQL 17 fixtures and isolated PAPERCLIP_HOME keep the run separate. All eight source hashes remain exact; no assertion, timeout, or source adaptation. - **Static gates pass:** database build; server typecheck/build; module boundaries; migration order; profile identity. All 412 migration journal entries are present through index/tag 0413_tired_kronos. - **Composition:** 1,237 parent-only and eight owned-only paths remain exact; 11,521 identical paths and two automatic shared production results are preserved. The only manual conflict is the status ledger. Packages and other native worker services equal the parent. Its 4,192 selected cases are prior source-scoped evidence, not added to this count. - **Historical failures remain recorded:** earlier trusted-runner and Greptile service errors are not current passes. The older Ponytail Ship verdict names another head. Fresh CI, security, and exact-head Greptile 5/5 must pass before merge. - [Master integration and spool acceptance audit](https://github.com/paperclipai/paperclip/blob/016b3af51a734ad0ddf574de701679a7bdc33d06/doc/plans/2026-10-26-import-spool-cleanup-sleep.md#current-master-integration-and-spool-acceptance-2026-10-29). No full local repository, live provider, or deployed sleep claim is made.
Earlier revision verification and failure history Receiver failure fairness: a controlled 1,100 ms batch reproduced a 30-second delay for healthy due receivers. The worker now supplies its existing per-item deadline through `nextRunAtOnError`. All 20 receiver lifecycle tests pass on the owner and combined tip `31c12f1d68ff4a51e06fc2be2abf2c2aee9f9d3a`; server production typechecking passes on the owner. The failed receiver retains its own backoff, and settled receivers return to zero recurring SQL. The five-second assertion budget is unchanged. See [the owning audit](https://github.com/paperclipai/paperclip/blob/7435a19f6d18f14466175d72b3b7d893ed39acd0/doc/plans/2026-10-25-chat-receiver-discovery-sleep-audit.md#healthy-receiver-progress-after-a-slow-failed-batch). Current-head CI and review remain required. Approved delegated Test-tab calls now evaluate caller and target policies in the existing claim transaction. Real HTTP regressions first showed a missing caller charge and shared-limit self-rejection. All 78 connection-management cases and 92 gateway cases now pass on the owning branch, along with server production typechecking. All 78 connection-management cases also pass at combined tip `d25f830db89c1623bff3547bfc68015c1c66e0cf`. The three new cases cover shared charges, separate charges, rollback on denial, duplicate approval, and rejection of a later authorized call. Intermediate counter/identity fixture failures and a reused-database Slack collision are recorded in [the owning audit](https://github.com/paperclipai/paperclip/blob/1509a56a97f2ffb2ab59a9a27f6d7be16ab46569/doc/idle-approved-tool-recovery.md#delegated-test-tab-admission-after-policy-changes). The earlier combined checks below remain evidence for their stated revisions. Current-head CI and review remain required. Inherited voice-cleanup verification: the controlled slow-provider regression and all 13 lifecycle tests passed on its owner and combined tip `ab22d8ed0c09028315607414b83b375e5cc0a60a`. That repair changed only the fixture clock advancement and its audit note. Current head: `7435a19f6d18f14466175d72b3b7d893ed39acd0`. This remains a single-duty PR stacked on #17801. - Integrated pinned master `8548173fc9efdc0fa07644fb92dcc0b6851a95e5` with normal merges. Every branch passed generated-schema checks and its target-branch migration-order gate. Existing SQL is unchanged; the 25 stack migrations follow master’s new migration. - Duty-specific producer coverage, lifecycle tests, and earlier failure records: [2026-10-24-recurring-server-sleep-stack.md](https://github.com/paperclipai/paperclip/blob/47ba0a93c4b01484ad58d16bbeb5a372c76b5a0d/doc/plans/2026-10-24-recurring-server-sleep-stack.md), [2026-10-26-import-spool-cleanup-sleep.md](https://github.com/paperclipai/paperclip/blob/47ba0a93c4b01484ad58d16bbeb5a372c76b5a0d/doc/plans/2026-10-26-import-spool-cleanup-sleep.md). - Combined tip `51e6cbe67979b55b340ea9f0f434c5b5c682ccd6`: **355 startup/cancellation/recovery/dispatch/preservation tests**, **75 connection-management tests**, and production server typechecking pass. These combined checks do not mean every intermediate PR independently ran the full repository suite. - Real master-to-stack database upgrade passes: **379 unchanged prior migration receipts plus 25 new migrations**, with the gateway responsibility fixture preserved. - Owning-branch checks: 84 runner installation/settlement cases plus runner typechecking and generated TypeScript/Rust profile agreement; 253 GitHub review workflow cases. Stop-attribution and held-worker readiness regressions failed before their fixes. - Failure accounting: the initial combined run passed 349 cases and failed six idle assertions in a reused preservation database containing old cancelled runs. Repeating the combined command with a fresh preservation database passed. Earlier invalid project selection, merge import/type errors, and the interrupted full local suite are not counted as passing checks. - **Latest-head CI and review remain required.** Prior CI findings include GitHub idle observation, route teardown, attention cleanup, touch-picker E2E, workspace health, and voice-cleanup timing; local passes do not erase those failures. No full local repository-suite, deployed sleep, fleet cost, or live Copilot qualification claim is made.
## Risks - Expiry and active apply must keep their guarded ledger claims. A stale read must never authorize file deletion after a concurrent resume. - Database and permission failures remain unresolved work. They cannot be interpreted as missing rows or empty directories. - Only supported ledger and spool owner writers emit these local notifications. Independent processes or SQL writers need ingress; startup recovery alone cannot wake stopped compute. - Cloud sleep blockers remain. No deployed suspension, external alarm coverage, live provider qualification, or cost savings is claimed. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, code execution, real PostgreSQL and filesystem tests, and GitHub tools. The exact runtime model identifier and context-window size are not exposed in this session. ## 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 URLs) - [x] My branch name describes the change 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 - [ ] All Paperclip CI gates are green - [ ] 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 Co-authored-by: Codex --- ...2026-10-24-recurring-server-sleep-stack.md | 18 +- .../2026-10-26-import-spool-cleanup-sleep.md | 263 ++++++++++++++++ .../company-import-transfer-routes.test.ts | 57 +++- .../__tests__/company-transfer-runs.test.ts | 13 +- .../import-spool-cleanup-work.test.ts | 283 ++++++++++++++++++ .../src/__tests__/tool-access-service.test.ts | 137 +++++---- server/src/app.ts | 46 +-- server/src/routes/companies.ts | 1 + .../src/services/company-import-transfers.ts | 158 +++++----- server/src/services/company-transfer-runs.ts | 75 +++-- .../services/delivery-work-notifications.ts | 1 + .../src/services/import-spool-cleanup-work.ts | 56 ++++ 12 files changed, 899 insertions(+), 209 deletions(-) create mode 100644 doc/plans/2026-10-26-import-spool-cleanup-sleep.md create mode 100644 server/src/__tests__/import-spool-cleanup-work.test.ts create mode 100644 server/src/services/import-spool-cleanup-work.ts 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 }; +}