From 59822467710a7c968347e3d8fe905359a348df2a Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 27 Oct 2026 14:17:01 -0500 Subject: [PATCH] fix(server): schedule custom-image setup expiry from committed changes (#17443) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip manages AI agent work and its execution environments. > - Interactive image setup sessions must expire even when users leave. > - The server currently scans these sessions on every maintenance tick. > - Empty scans keep the instance and database active. > - Existing session rows already record status and expiry. > - This pull request uses transaction signals and the shared scheduler for those rows. > - Empty and completed setup history no longer create recurring cleanup queries. ## Linked Issues or Issue Description **What happened?** The server checks custom-image setup expiry every heartbeat maintenance tick, even when no active session exists. **What did you expect?** Session writes should update the next cleanup deadline. Existing provider cancellation and cleanup ownership must remain intact. Empty state should stay quiet. **Steps to reproduce** Run an instance with heartbeat scheduling enabled and no custom-image setup sessions. Observe repeated setup-expiry reads. Refs #17145. This is the first PR in the recurring server sleep stack. Its base is master. The full requested scope and verification contract are recorded in `doc/plans/2026-10-24-recurring-server-sleep-stack.md`. Follow-up PRs target their predecessor and each own a separate reason for recurring work. ## What Changed - Carry the historical cleanup retry-observation fixture repair and its existing external test-database hook at the stack foundation. Runtime scheduling is unchanged. - Carry the previously verified login-cleanup fixture settlement correction at the stack foundation so every intermediate CI run receives it. This is test-only maintenance; custom-image expiry behavior is unchanged. - Add a custom-image expiry worker to the existing delivery coordinator. Remove its maintenance-tick scan. - Signal initial creation, provider refresh, cancellation/failure, and promotion inside their owning transaction. - Keep the active-status predicate, earliest saved expiry, and provider behavior in the custom-image service. Shared code gains only a queue topic. - Preserve startup recovery, bounded retries, concurrent write generations, and drain deferral. No tables or external alarm changes. - Add PostgreSQL lifecycle tests and repair this suite's database cleanup callback. A dedicated external test URL permits disposable PostgreSQL when the host cannot start another embedded cluster. ## Verification Current head: `3507c6be1e054fdd93353d5dcf77d4899b44518f`. This remains a single-duty PR stacked on master. - 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-08-instance-sleep-and-scheduler-inventory.md](https://github.com/paperclipai/paperclip/blob/3507c6be1e054fdd93353d5dcf77d4899b44518f/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md), [2026-10-24-recurring-server-sleep-stack.md](https://github.com/paperclipai/paperclip/blob/3507c6be1e054fdd93353d5dcf77d4899b44518f/doc/plans/2026-10-24-recurring-server-sleep-stack.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 - Signals cover supported service producers sharing the DB owner. Independent SQL and another process require the existing restart/reconciliation contract. - Failed provider cancellation retains the existing failed-session and historical-cleanup behavior. This PR changes scheduling, not teardown authority. - Active rows without an expiry preserve the existing behavior; no deadline is invented. Cloud sleep eligibility remains conservative. - The worker remains in the existing heartbeat-enabled startup path. It does not change disabled-scheduler behavior or prove deployed suspension. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact serving model identifier and context window were not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] 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 Review follow-up: Added a real producer commit during a paused deadline read. Updated startup assertions for coordinator ownership and removed added code comments. The three affected suites pass (102 tests). The separate login-cleanup suite also passes locally (16 tests); its prior CI consume test timed out, and the fresh complete CI run passed. External PostgreSQL tests now use the repository guard for disposable databases ending in `_test`. --------- Co-authored-by: Paperclip --- doc/idle-environment-cleanup.md | 45 ++++ doc/idle-login-cleanup.md | 17 ++ ...-instance-sleep-and-scheduler-inventory.md | 9 +- ...2026-10-24-recurring-server-sleep-stack.md | 63 ++++++ ...-10-26-retained-workspace-slot-fairness.md | 53 +++++ packages/db/src/migrations/meta/_journal.json | 2 +- .../tests/native_provider_backend.rs | 2 +- .../custom-image-cleanup-work.test.ts | 75 +++++++ .../environment-cleanup-work.test.ts | 16 +- .../environment-custom-images-service.test.ts | 199 +++++++++++++++++- .../src/__tests__/login-cleanup-work.test.ts | 8 +- .../server-startup-feedback-export.test.ts | 12 +- server/src/index.ts | 22 +- .../src/services/custom-image-cleanup-work.ts | 35 +++ .../services/delivery-work-notifications.ts | 1 + .../src/services/environment-custom-images.ts | 33 ++- .../src/services/workspace-recovery.test.ts | 22 +- server/src/services/workspace-recovery.ts | 16 +- 18 files changed, 569 insertions(+), 61 deletions(-) create mode 100644 doc/plans/2026-10-24-recurring-server-sleep-stack.md create mode 100644 doc/plans/2026-10-26-retained-workspace-slot-fairness.md create mode 100644 server/src/__tests__/custom-image-cleanup-work.test.ts create mode 100644 server/src/services/custom-image-cleanup-work.ts diff --git a/doc/idle-environment-cleanup.md b/doc/idle-environment-cleanup.md index 9070703ef8c..a8d2ecdff02 100644 --- a/doc/idle-environment-cleanup.md +++ b/doc/idle-environment-cleanup.md @@ -101,3 +101,48 @@ moves here. Other maintenance duties, external probes, and connected transports still affect idle traffic. No Cloud alarm coverage or host-sleep exemption changes. These tests prove the changed workers become quiet and recover; they do not prove deployed suspension or fleet savings. + +## Custom-image setup expiry + +Setup expiry uses a separate `custom-image-expiry` worker. The custom-image +service owns the active-status predicate, the earliest saved `expiresAt`, and +provider cancellation. Startup registers the worker in the existing enabled +heartbeat housekeeping path; the maintenance tick no longer calls its sweep. + +All supported setup producers live in `environment-custom-images.ts`: initial +creation, provider refresh, cancellation/failure status, and promotion. They +register intent in their owning transaction. A provider refresh can move the +deadline earlier; a terminal transition removes it. Historical verification +metadata does not change active expiry eligibility and does not need this signal. +Environment deletion can only remove obligations; an already scheduled pass +observes the removal. Independent SQL and other DB roots are outside the existing +process-local notification contract. + +No active dated session means no recurring reads or provider calls. Overdue +batches and failures retain a bounded retry, while first writes and failed +reconciliation survive drain. A missing expiry retains the existing behavior +(no inferred expiration); sleep safety still treats unsupported live resources +conservatively. Provider cancellation, failure state, and historical resource +verification keep their existing semantics. No Cloud alarm coverage changes. + +## Historical retry test observation + +The historical verification worker intentionally does not delay server readiness. +Its provider sweep finishes before the following database pending-state read +registers a retry deadline. The regression test now waits for that future +deadline before inspecting it; sweep completion alone is not scheduler settlement. +The runtime worker, retry interval, and readiness contract are unchanged. + +A controlled 150ms delay before the real pending-state query reproduced the +current CI failure: the immediate assertion observed a null deadline. The +corrected assertion passed with the same delay. The temporary delay was removed, +and all 14 cleanup tests passed against an isolated PostgreSQL database. The +existing external test-database hook is carried at the stack foundation so +intermediate branches can use this same verification path. + +An initial attempt at the foundation could not start embedded PostgreSQL because +that earlier branch lacked the external fixture hook. A broader fixture wait for +all idle holds to clear then failed three cases that legitimately retain pending +cleanup obligations. That wait was removed. Only the historical deadline assertion +changed; its requirement is now a deadline strictly in the future. The complete +suite passes with the original readiness and idle-hold behavior preserved. diff --git a/doc/idle-login-cleanup.md b/doc/idle-login-cleanup.md index 813cf34eb85..5bc3ebf62ac 100644 --- a/doc/idle-login-cleanup.md +++ b/doc/idle-login-cleanup.md @@ -95,3 +95,20 @@ cover the surrounding login contracts. The fake-clock database fixture warms its connections and disables pool idle/lifetime expiry so time jumps test the worker rather than racing socket connection timers. Database suspension and reconnect behavior require separate qualification. + +## CI fixture settlement across the migration stack + +The setup-token consume fixture can finish its initial fake-clock jump while +cleanup is still performing database work. A write during that pass can request +another generation whose coalesced deadline lies beyond the jump. Waiting for +zero tracked work without advancing to that registered deadline then times out, +even though cleanup has a valid wake and can finish. + +The existing fixture repair from `59f8c8b3e5` is now carried at the earliest stack +branch so every intermediate PR receives it. The helper advances to registered +deadlines only while work remains tracked, preserving its existing five-second +observation limit and all production behavior. All 16 login-cleanup cases passed +on this branch against a dedicated PostgreSQL database; the prior later-branch +verification and failed CI history remain documented in the GitHub review-check +migration audit. No runtime change or weakened quiet assertion is included. +Evidence: `/tmp/sleep-login-settlement-owning-tests.log` and its zero status file. diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md index 2e5bb5c9ca5..b92d7520ec0 100644 --- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md +++ b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md @@ -634,10 +634,11 @@ Longer polling intervals alone do not complete an item. [idle-login-cleanup.md](../idle-login-cleanup.md) for producer coverage, verification and remaining provider-orphan boundaries. -- [ ] **JOB-19 — Custom-image setup expiry.** Current: **30s** active-session - expiry scans. Change: register `expiresAt` and cleanup attempts. **Complete - when:** no sessions means no SQL and expired provider resources are released. - Source: [environment-custom-images.ts](../../server/src/services/environment-custom-images.ts). +- [x] **JOB-19 — Custom-image setup expiry.** Uses committed setup-session + changes and the earliest existing active-session expiry. Startup reconstructs + deadlines; empty and terminal history disarm. Failed reconciliation and first + writes retain their wake through drain. Provider teardown and historical cleanup + remain with the existing services. See [idle-environment-cleanup.md](../idle-environment-cleanup.md#custom-image-setup-expiry). - [ ] **JOB-20 — Tool connection health.** Current: **30s** DB inventory; provider probes are conditional. Change: check on use/configuration change, diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md new file mode 100644 index 00000000000..50f73222265 --- /dev/null +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -0,0 +1,63 @@ +# Recurring server work migration stack + +Baseline: `785d54636e1c8513ee5503d602daf6a5ee1970d6` (2026-10-24). + +The requested outcome is a reviewed PR stack for every remaining recurring server +work item below. A PR owns one reason for background activity. Split compound +items into separate PRs as their producer and lifecycle audits identify distinct +owners. Each PR includes behavior, producer audit, tests, and documentation. +Do not mark an item done from a timer removal or a green check alone. + +## Scope and completion ledger + +| ID | Original checklist duty | Required slices / disposition | PR / evidence | +| --- | --- | --- | --- | +| 01 | Custom-image setup expiry | Session writes, existing expiry, startup/retry/drain | In progress | +| 02 | Remaining chat maintenance | Provider runtime discovery; voice reconciliation; milestones; wake notices; interaction continuations; terminal confirmations; GitHub recovery; Teams maintenance, separately by owner | Pending | +| 03 | Run and task recovery | Orphan runs and active leases; shutdown workspace settlement; scheduled retries; queued runs; stranded assignments; resolved dependencies; task watchdogs; silent-run monitoring; stale locks | Pending | +| 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Pending | +| 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending | +| 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending | +| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Pending | +| 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending | +| 09 | Merged-PR confirmations | Candidate writes, webhook activation where supported, outstanding-candidate fallback | Pending | +| 10 | External-object refresh | Configuration changes, refresh deadlines, source events | Pending | +| 11 | Status-card evaluation | Settings/card mutations and evaluation deadlines | Pending | +| 12 | Plugin scheduled jobs | Registration/update/removal and actual job deadlines | Pending | +| 13 | Routine, agent-timer and issue-monitor scheduling | Remove periodic wake backstops after complete producer coverage; routine execution-status reconciliation separately | Pending | +| 14 | Skill-version retention and native-harness backup expiration | Separate retention owners and existing retention guarantees | Pending | +| 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 | + +## Stack and verification contract + +The first PR targets master. Each following PR targets its predecessor branch so +its diff shows only its own reason. Parent changes must be propagated before +final verification. Record the exact checked head and base for every PR. All PRs +must finish with passing checks, fresh 5/5 Greptile review, and no unresolved +findings. This request does not authorize merging the stack. + +Reuse the delivery coordinator, transaction notifications, and shared scheduler. +Existing durable rows own work; no table per loop. Keep domain/provider behavior +in its service and wiring in startup. A write signal is not durable truth. Preserve +outer commit/rollback and uncertain-outcome reconciliation. Audit every supported +producer, including alternate routes and recovery writers. + +For each slice verify applicable empty/configured-idle/history quiet states, +first real producer, actual deadlines, rollback/outer commit, restart, failures, +changes during execution, drain/resume, and shutdown. Use real PostgreSQL where +transaction boundaries matter and controlled providers where external behavior +is under test. Record failures and subsequent repair evidence. + +External-source reconciliation may remain only for a concrete supported obligation +with a defined cadence and stopping condition, using the shared scheduler. Never +remove the only missed-event backstop without replacing its recovery contract. + +## Boundaries + +Cloud sleep eligibility and durable alarm coverage remain conservative. An +in-process deadline cannot wake a stopped host. Browser traffic, persistent +connector transport architecture, probes, telemetry/exporter qualification, and +end-to-end provider suspension are outside this requested recurring-server stack; +none are claimed solved by these PRs. Required active-operation lease renewals and +request timeouts remain with their owners. diff --git a/doc/plans/2026-10-26-retained-workspace-slot-fairness.md b/doc/plans/2026-10-26-retained-workspace-slot-fairness.md new file mode 100644 index 00000000000..682bf991baa --- /dev/null +++ b/doc/plans/2026-10-26-retained-workspace-slot-fairness.md @@ -0,0 +1,53 @@ +# Retained workspace recovery: fill a free slot across the cursor + +This is a shared CI repair carried at the sleep-stack root. It fixes the existing +retained-workspace owner; it does not add another background duty or migrate the +remaining run/task recovery inventory. No PR is merged by this repair. + +## Failure and ownership + +The retained-workspace worker permits two physical exports. A held export stays +in the durable unfinished candidates, even though this process already owns it. +Discovery paginates by recovery-action ID and wraps only when its query returns +no rows. A page containing only a held export was nonempty, but its only row was +skipped in memory. Newly committed work before the cursor therefore waited for +another 15-second recovery pass despite a free slot. + +The CI failure at PR #17446, run 40423925338 / job 121925042581, leaves the third +export `scheduled` at the existing 20-second assertion. The older #17450 failure +has the same observable result. A deterministic real-PostgreSQL/filesystem +regression reproduces it by starting the highest action ID, holding its provider +resume, completing a middle action, then requesting the lowest action. The +before-fix test failed at that same third-export assertion. + +## Change and preserved boundaries + +The existing candidate query accepts the locally in-flight run IDs and excludes +them before pagination on both unfinished and passive discovery paths. The +coordinator reads this set after its existing two-slot guard, so the exclusion +contains at most one run for an admitted discovery pass. An empty page can now +wrap in the same pass and find new work before the cursor. The in-memory duplicate +guard remains to handle changes while a query is running. + +The retained-workspace module owns candidate selection. Its callers, SQL source +eligibility, physical ownership lock, retry policy, concurrency limit, completion +callbacks, and shutdown joins retain their existing contracts. Other-process +locks are still handled by the existing bounded cursor rotation. No new timer, +queue, table, provider branch, or retry authority is added. Held work remains +tracked as active and is not treated as settled or safe to sleep through. + +## Verification + +At the integrated pre-repair tip, the deterministic test failed with `scheduled` +instead of `complete`. After the repair, that regression and the existing +other-worker physical-lock rotation test both pass with real files and database +state. The free-slot test keeps the original 20-second completion bound, verifies +preserved files, no extra agent run, at most two concurrent resume operations, +and shutdown waiting for the held export before its final completion. + +At the stack root, all 56 retained-workspace recovery cases pass on a dedicated +migrated PostgreSQL test database. The full suite also exercises retries, +exhaustion, stop verification, cleanup authorization, source bindings and saved +file recovery. Integrated workspace typecheck/build and fresh remote checks must +be recorded in the PR after parent propagation. This is not a full repository or +deployed-sleep claim. diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index b6d2fe1f6bd..96c505941fd 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2656,4 +2656,4 @@ "breakpoints": true } ] -} \ No newline at end of file +} diff --git a/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs b/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs index 2e99a0bb35c..eebe8fb4c16 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/tests/native_provider_backend.rs @@ -1160,7 +1160,7 @@ fn acpx_transport_failure_retains_its_error_without_reopening_the_provider() { assert_eq!(snapshot["status"], "closed"); assert_eq!(snapshot["activeProviderTurnId"], Value::Null); assert!(executor.poll_events().unwrap().is_empty()); - executor.shutdown().unwrap(); + shutdown_recovered_acpx_fixture(&mut executor, &directory); fs::remove_dir_all(directory).unwrap(); } diff --git a/server/src/__tests__/custom-image-cleanup-work.test.ts b/server/src/__tests__/custom-image-cleanup-work.test.ts new file mode 100644 index 00000000000..6a6b5b021d9 --- /dev/null +++ b/server/src/__tests__/custom-image-cleanup-work.test.ts @@ -0,0 +1,75 @@ +import { createDb } from "@paperclipai/db"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { registerCustomImageCleanupWork } from "../services/custom-image-cleanup-work.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; + +describe("custom-image expiry deadline precision", () => { + let db: ReturnType; + let coordinator: ReturnType; + beforeEach(() => { vi.useFakeTimers(); vi.setSystemTime(new Date("2026-10-26T12:00:00Z")); + db = createDb("postgres://localhost:1/deadline_precision_test"); }); + afterEach(async () => { await coordinator?.stop(); await db.$client.end({ timeout: 1 }); vi.useRealTimers(); }); + + async function start(delay: number, failFirst = false) { + let due: number | null = Date.now() + delay; + let expired = 0; + const errors: unknown[] = []; + const cleanup = vi.fn(async () => { + if (due !== null && due <= Date.now()) { expired++; due = null; } + return { scanned: 0, timedOut: 0, failed: 0 }; + }); + if (failFirst) cleanup.mockRejectedValueOnce(new Error("temporary database failure")); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, + onError: error => errors.push(error) }); + const worker = registerCustomImageCleanupWork(coordinator, { + cleanupExpiredSetupSessions: cleanup, + nextSetupCleanupAt: async () => due, + }, () => true); + await worker.ready; + await vi.advanceTimersByTimeAsync(0); + return { cleanup, errors, worker, expired: () => expired, move: (at: number) => { due = at; worker.wake(); } }; + } + + it("expires at a saved five-second startup deadline and then becomes quiet", async () => { + const due = Date.now() + 5_000; + const work = await start(5_000); + expect(coordinator.nextWakeAt()).toBe(due); + await vi.advanceTimersByTimeAsync(4_999); + expect(work.expired()).toBe(0); + await vi.advanceTimersByTimeAsync(1); + expect(work.expired()).toBe(1); + expect(coordinator.nextWakeAt()).toBeNull(); + const calls = work.cleanup.mock.calls.length; + await vi.advanceTimersByTimeAsync(120_000); + expect(work.cleanup).toHaveBeenCalledTimes(calls); + }); + + it("replaces a future deadline with an earlier wake inside the retry window", async () => { + const work = await start(60_000); + await vi.advanceTimersByTimeAsync(1_000); + const due = Date.now() + 5_000; + work.move(due); + await vi.advanceTimersByTimeAsync(0); + expect(coordinator.nextWakeAt()).toBe(due); + await vi.advanceTimersByTimeAsync(4_999); + expect(work.expired()).toBe(0); + await vi.advanceTimersByTimeAsync(1); + expect(work.expired()).toBe(1); + }); + + it("retains failure retry spacing even when another wake arrives", async () => { + const retryAt = Date.now() + 30_000; + const work = await start(5_000, true); + expect(work.errors).toHaveLength(1); + work.worker.wake(); + await vi.advanceTimersByTimeAsync(0); + expect(coordinator.nextWakeAt()).toBe(retryAt); + await vi.advanceTimersByTimeAsync(29_999); + expect(work.cleanup).toHaveBeenCalledTimes(1); + expect(work.expired()).toBe(0); + await vi.advanceTimersByTimeAsync(1); + expect(work.expired()).toBe(1); + expect(work.cleanup).toHaveBeenCalledTimes(2); + expect(coordinator.nextWakeAt()).toBeNull(); + }); +}); diff --git a/server/src/__tests__/environment-cleanup-work.test.ts b/server/src/__tests__/environment-cleanup-work.test.ts index 7f6b1d64c12..6a251dabdd0 100644 --- a/server/src/__tests__/environment-cleanup-work.test.ts +++ b/server/src/__tests__/environment-cleanup-work.test.ts @@ -17,17 +17,20 @@ import { historicalCleanupVerificationService } from "../services/historical-cle import { heartbeatService } from "../services/heartbeat.js"; import { environmentRuntimeService } from "../services/environment-runtime.js"; -const support = await getEmbeddedPostgresTestSupport(); +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_ENVIRONMENT_CLEANUP_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); describe.skipIf(!support.supported)("environment cleanup scheduling", () => { - let database: Awaited>, db: ReturnType; + let database: Awaited> | undefined, db: ReturnType; let companyId: string, environmentId: string, draining: boolean, failing: boolean; let coordinator: ReturnType | undefined; let worker: ReturnType | undefined; let releases: ReturnType, inspections: ReturnType, reads: ReturnType; const errors: unknown[] = [], disposers: Array<() => void> = []; beforeAll(async () => { - database = await startEmbeddedPostgresTestDatabase("environment-cleanup-work-"); - db = createDb(database.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + if (!external) database = await startEmbeddedPostgresTestDatabase("environment-cleanup-work-"); + db = createDb(external ?? database!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); }, 90_000); beforeEach(async () => { @@ -230,7 +233,10 @@ describe.skipIf(!support.supported)("environment cleanup scheduling", () => { it("keeps unknown historical resources scheduled until provider confirmation", async () => { const lease = await pending(); await environmentService(db).releaseLease(lease.id, "failed", { cleanupStatus: "success" }); inspections.mockResolvedValueOnce({ providerLeaseId: lease.providerLeaseId, state: "unknown" }); - await start(true); expect(coordinator!.nextWakeAt()).not.toBeNull(); + await start(true); + // The provider sweep finishes before its pending-state read registers the + // retry. Readiness intentionally does not wait for historical verification. + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBeGreaterThan(Date.now())); await settle(300_000); expect(inspections).toHaveBeenCalledTimes(2); await quiet(); }); }); diff --git a/server/src/__tests__/environment-custom-images-service.test.ts b/server/src/__tests__/environment-custom-images-service.test.ts index 6422cec2f91..d4cc612963d 100644 --- a/server/src/__tests__/environment-custom-images-service.test.ts +++ b/server/src/__tests__/environment-custom-images-service.test.ts @@ -1,6 +1,10 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { registerCustomImageCleanupWork } from "../services/custom-image-cleanup-work.js"; +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; import { randomUUID } from "node:crypto"; -import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; -import { desc, eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { desc, eq, sql } from "drizzle-orm"; import { activityLog, companies, @@ -27,7 +31,8 @@ import { startEmbeddedPostgresTestDatabase, } from "./helpers/embedded-postgres.js"; -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalTestDatabase = resolveExternalTestDatabaseUrl("PAPERCLIP_CUSTOM_IMAGE_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalTestDatabase ? { supported: true, reason: undefined } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; if (!embeddedPostgresSupport.supported) { @@ -153,9 +158,10 @@ describeEmbeddedPostgres("environmentCustomImageService", () => { let db!: ReturnType; beforeAll(async () => { + if (externalTestDatabase) { db = createDb(externalTestDatabase, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); return; } const started = await startEmbeddedPostgresTestDatabase("environment-custom-images"); - stopDb = started.stop; - db = createDb(started.connectionString); + stopDb = started.cleanup; + db = createDb(started.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); }); afterEach(async () => { @@ -167,6 +173,7 @@ describeEmbeddedPostgres("environmentCustomImageService", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await stopDb?.(); }); @@ -200,6 +207,176 @@ describeEmbeddedPostgres("environmentCustomImageService", () => { return { companyId, environmentId }; } + describe("expiry scheduling", () => { + let coordinator: ReturnType | undefined; + let service: ReturnType; + let provider: ReturnType; + let draining: boolean; + const errors: unknown[] = []; + + beforeEach(async () => { + await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`))); + draining = false; + errors.length = 0; + provider = createWorkerManager(); + service = environmentCustomImageService(db, { pluginWorkerManager: provider }); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); + coordinator = undefined; + vi.useRealTimers(); + vi.restoreAllMocks(); + expect(errors).toEqual([]); + }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + await registerCustomImageCleanupWork(coordinator, service, () => !draining).ready; + } + async function settle(ms = 60_000) { + await vi.advanceTimersByTimeAsync(ms); + if (!draining) await vi.waitFor(() => expect(idleWorkSnapshot().active === 0 || coordinator!.nextWakeAt() !== null).toBe(true), { timeout: 5000 }); + } + async function quiet() { + await settle(); + expect(coordinator!.nextWakeAt()).toBeNull(); + const reads = vi.spyOn(db, "select"), calls = provider.call.mock.calls.length; + await settle(30 * 60_000); + expect(reads).not.toHaveBeenCalled(); + expect(provider.call).toHaveBeenCalledTimes(calls); + reads.mockRestore(); + } + async function create() { + const fixture = await seed(); + return service.startSetupSession({ environmentId: fixture.environmentId, + actor: { userId: "fixture-user" }, ttlSeconds: 3600 }); + } + const cancellations = () => provider.call.mock.calls.filter(([, method]) => method === "environmentCancelInteractiveSetup"); + + it("is quiet when empty, wakes from the real producer, expires once, and keeps terminal history quiet", async () => { + await start(); await quiet(); + const created = await create(); + await settle(); + expect(coordinator!.nextWakeAt()).toBe(new Date(created.session.expiresAt!).getTime()); + const reads = vi.spyOn(db, "select"); + await settle(30 * 60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect(cancellations()).toHaveLength(0); + await settle(30 * 60_000); + expect(cancellations()).toHaveLength(1); + expect(await service.getSessionById(created.session.id)).toMatchObject({ status: "timed_out" }); + await quiet(); + }); + + it("honors a committed expiry inside the retry window", async () => { + await start(); + const created = await create(); + await settle(); + const earlier = new Date(Date.now() + 5_000); + provider.call.mockResolvedValueOnce({ providerLeaseId: created.session.providerLeaseId, + status: "waiting_for_user", expiresAt: earlier.toISOString() } as never); + await service.refreshSetupSession({ sessionId: created.session.id }); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(earlier.getTime())); + expect(cancellations()).toHaveLength(0); + await settle(Math.max(0, earlier.getTime() - Date.now())); + await vi.waitFor(() => expect(cancellations()).toHaveLength(1)); + expect(await service.getSessionById(created.session.id)).toMatchObject({ status: "timed_out" }); + await quiet(); + }); + + it("replaces an expiry when the provider moves it earlier and disarms on user cancellation", async () => { + await start(); + const created = await create(); await settle(); + const earlier = new Date(Date.now() + 5 * 60_000); + provider.call.mockResolvedValueOnce({ providerLeaseId: created.session.providerLeaseId, + status: "waiting_for_user", expiresAt: earlier.toISOString() } as never); + await service.refreshSetupSession({ sessionId: created.session.id }); + await settle(); + expect(coordinator!.nextWakeAt()).toBe(earlier.getTime()); + await service.cancelSetupSession({ sessionId: created.session.id }); + await quiet(); + expect(cancellations()).toHaveLength(1); + }); + + it.each([false, true])("waits for the producer's outer transaction (rollback %s)", async rollback => { + const fixture = await seed(); await start(); await quiet(); + let release!: () => void, written!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const staged = new Promise(resolve => { written = resolve; }); + const writing = db.transaction(async tx => { + const scoped = environmentCustomImageService(tx as unknown as typeof db, { pluginWorkerManager: provider }); + await scoped.startSetupSession({ environmentId: fixture.environmentId, actor: { userId: "fixture-user" }, ttlSeconds: 1 }); + written(); await gate; + if (rollback) throw new Error("rollback fixture"); + }).catch(error => { if (!rollback) throw error; }); + await staged; + try { + await vi.advanceTimersByTimeAsync(60_000); + expect(cancellations()).toHaveLength(0); + } finally { release(); } + await writing; await settle(); + expect(cancellations()).toHaveLength(rollback ? 0 : 1); + await quiet(); + }); + + it("retains a producer commit that arrives during the deadline read", async () => { + await start(); const created = await create(); await settle(); + await coordinator!.stop(); + let release!: () => void, reading!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const entered = new Promise(resolve => { reading = resolve; }); + const readDeadline = service.nextSetupCleanupAt; + vi.spyOn(service, "nextSetupCleanupAt").mockImplementationOnce(async () => { + const stale = await readDeadline(); + reading(); await gate; + return stale; + }); + const starting = start(); + await entered; + const earlier = new Date(Date.now() + 5 * 60_000); + try { + provider.call.mockResolvedValueOnce({ providerLeaseId: created.session.providerLeaseId, + status: "waiting_for_user", expiresAt: earlier.toISOString() } as never); + await service.refreshSetupSession({ sessionId: created.session.id }); + } finally { release(); } + await starting; await settle(); + await vi.waitFor(() => expect(coordinator!.nextWakeAt()).toBe(earlier.getTime())); + await settle(5 * 60_000); + await vi.waitFor(() => expect(cancellations()).toHaveLength(1)); + await quiet(); + }); + + it("reconstructs a saved expiry after restart", async () => { + await start(); const created = await create(); await settle(); + await coordinator!.stop(); + await start(); + expect(coordinator!.nextWakeAt()).toBe(new Date(created.session.expiresAt!).getTime()); + await settle(60 * 60_000); + expect(cancellations()).toHaveLength(1); await quiet(); + }); + + it.each(["startup", "first_write", "failed_pass"])("retains %s work through drain without passive SQL", async mode => { + if (mode === "startup") draining = true; + await start(); + if (mode === "first_write") { await quiet(); draining = true; } + await create(); + if (mode === "failed_pass") { + vi.spyOn(service, "nextSetupCleanupAt").mockRejectedValueOnce(new Error("database unavailable")); + await settle(30_000); + await vi.waitFor(() => expect(errors).toHaveLength(1)); errors.length = 0; + draining = true; + } + await settle(); + const reads = vi.spyOn(db, "select"); + await settle(2 * 60 * 60_000); + expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect(cancellations()).toHaveLength(0); + draining = false; await settle(); + expect(cancellations()).toHaveLength(1); await quiet(); + }); + }); + it.each(["supported", "old_worker", "changed_config"])("inspects historical setup read-only with exact config and capability gates: %s", async mode => { const fixture = await seed(); const worker = createWorkerManager({ workerMethods: [...ALL_CUSTOM_IMAGE_WORKER_METHODS, @@ -838,9 +1015,10 @@ describeEmbeddedPostgres("environmentCustomImageService reconciliation", () => { let db!: ReturnType; beforeAll(async () => { + if (externalTestDatabase) { db = createDb(externalTestDatabase, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); return; } const started = await startEmbeddedPostgresTestDatabase("environment-custom-images-reconcile"); - stopDb = started.stop; - db = createDb(started.connectionString); + stopDb = started.cleanup; + db = createDb(started.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); }); afterEach(async () => { @@ -852,6 +1030,7 @@ describeEmbeddedPostgres("environmentCustomImageService reconciliation", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await stopDb?.(); }); @@ -1145,9 +1324,10 @@ describeEmbeddedPostgres("environmentCustomImageService relink", () => { let db!: ReturnType; beforeAll(async () => { + if (externalTestDatabase) { db = createDb(externalTestDatabase, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); return; } const started = await startEmbeddedPostgresTestDatabase("environment-custom-images-relink"); - stopDb = started.stop; - db = createDb(started.connectionString); + stopDb = started.cleanup; + db = createDb(started.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); }); afterEach(async () => { @@ -1160,6 +1340,7 @@ describeEmbeddedPostgres("environmentCustomImageService relink", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await stopDb?.(); }); diff --git a/server/src/__tests__/login-cleanup-work.test.ts b/server/src/__tests__/login-cleanup-work.test.ts index c21d04c97d3..081fc6eece6 100644 --- a/server/src/__tests__/login-cleanup-work.test.ts +++ b/server/src/__tests__/login-cleanup-work.test.ts @@ -116,7 +116,13 @@ describe.skipIf(!support.supported)("durable login cleanup", () => { }); async function settle(ms = 30_000) { await vi.advanceTimersByTimeAsync(ms); - await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 }); + await vi.waitFor(async () => { + if (idleWorkSnapshot().active > 0) { + const deadlines = coordinators.flatMap(c => c.nextWakeAt() === null ? [] : [c.nextWakeAt()!]); + if (deadlines.length) await vi.advanceTimersByTimeAsync(Math.max(0, Math.min(...deadlines) - Date.now())); + } + expect(idleWorkSnapshot().active).toBe(0); + }, { timeout: 5000 }); await vi.advanceTimersByTimeAsync(1); await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0), { timeout: 5000 }); expect(errors).toEqual([]); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index db22ce7b2c2..81c689366c3 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -129,6 +129,7 @@ const { }; const issueThreadInteractionServiceFactoryMock = vi.fn(() => issueThreadInteractionServiceMock); const environmentCustomImagesServiceMock = { + nextSetupCleanupAt: vi.fn(async (): Promise => null), cleanupExpiredSetupSessions: vi.fn(async () => ({ scanned: 0, timedOut: 0, failed: 0 })), }; const environmentCustomImagesServiceFactoryMock = vi.fn(() => environmentCustomImagesServiceMock); @@ -843,7 +844,8 @@ describe("startServer feedback export wiring", () => { expect(issueThreadInteractionServiceMock.sweepMergedPullRequestConfirmations).toHaveBeenCalledTimes(1); expect(executionWorkspaceServiceMock.sweepTerminalWorkspaces).toHaveBeenCalledTimes(1); expect(routineServiceMock.tickScheduledTriggers).toHaveBeenCalledTimes(1); - expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(2); + expect(environmentCustomImagesServiceMock.cleanupExpiredSetupSessions).toHaveBeenCalledTimes(1); + expect(deliveryWorkRegisterMock.mock.calls.some(([queue]) => queue === DELIVERY_QUEUES.customImageExpiry)).toBe(true); } finally { setIntervalSpy.mockRestore(); } @@ -960,10 +962,10 @@ describe("startServer feedback export wiring", () => { try { await expect(startServer()).resolves.toBeDefined(); await settleStartupHousekeeping(); - expect(vi.mocked(logger.error)).toHaveBeenCalledWith( - expect.objectContaining({ startupStep: "custom_image_setup_cleanup" }), - "startup housekeeping step failed", - ); + const registration = deliveryWorkRegisterMock.mock.calls.find(([queue]) => queue === DELIVERY_QUEUES.customImageExpiry); + expect(registration).toBeDefined(); + expect(await registration![1].hasPending!()).toBe(true); + expect(environmentCustomImagesServiceMock.nextSetupCleanupAt).not.toHaveBeenCalled(); // The scheduler starts only after every housekeeping step settled. expect(heartbeatIntervals).toHaveLength(1); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index 23b13375283..411dda3f87f 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerCustomImageCleanupWork } from "./services/custom-image-cleanup-work.js"; import { registerWorkspaceRecoveryWork } from "./services/workspace-recovery-work.js"; import { registerEnvironmentCleanupWork } from "./services/environment-cleanup-work.js"; import { loginCleanupSchedule } from "./services/login-cleanup-state.js"; @@ -1790,12 +1791,9 @@ async function startServerWithDatabaseTeardown( work().catch((err: unknown) => { logger.error({ err, startupStep: name }, "startup housekeeping step failed"); }); - await step("custom_image_setup_cleanup", async () => { - const setupCleanup = await environmentCustomImages.cleanupExpiredSetupSessions(); - if (setupCleanup.timedOut > 0 || setupCleanup.failed > 0) { - logger.warn({ ...setupCleanup }, "startup environment customImage setup cleanup changed sessions"); - } - }); + await step("custom_image_setup_cleanup", () => registerCustomImageCleanupWork(deliveryWork, environmentCustomImages, + () => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped, + config.heartbeatSchedulerIntervalMs).ready); await step("tool_connection_health", async () => { const toolHealthSweep = await tools.sweepConnectionHealth(); if (toolHealthSweep.failed > 0) { @@ -1877,18 +1875,6 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "status-card scheduler tick failed"); })); - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; - trackHeartbeatSchedulerWork(environmentCustomImages - .cleanupExpiredSetupSessions() - .then((result) => { - if (result.timedOut > 0 || result.failed > 0) { - logger.warn({ ...result }, "environment customImage setup cleanup changed sessions"); - } - }) - .catch((err) => { - logger.error({ err }, "environment customImage setup cleanup failed"); - })); - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; trackHeartbeatSchedulerWork(tools .sweepConnectionHealth() diff --git a/server/src/services/custom-image-cleanup-work.ts b/server/src/services/custom-image-cleanup-work.ts new file mode 100644 index 00000000000..9ecd4e73612 --- /dev/null +++ b/server/src/services/custom-image-cleanup-work.ts @@ -0,0 +1,35 @@ +import { logger } from "../middleware/logger.js"; +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import type { environmentCustomImageService } from "./environment-custom-images.js"; + +export function registerCustomImageCleanupWork( + coordinator: ReturnType, + service: Pick, "cleanupExpiredSetupSessions" | "nextSetupCleanupAt">, + canRun: () => boolean, + retryMs = 30_000, +) { + let requested = 1, reconciled = 0; + let next: number | null = null; + let earliest = 0; + return coordinator.register(DELIVERY_QUEUES.customImageExpiry, { + retryMs, + pausePassiveMaintenance: true, + onSettled: () => { requested++; }, + run: async () => { + if (!canRun()) { requested++; return; } + if (Date.now() < earliest) { next = earliest; return; } + const generation = ++requested; + earliest = Date.now() + retryMs; + const result = await service.cleanupExpiredSetupSessions(); + if (result.timedOut > 0 || result.failed > 0) logger.warn(result, "environment customImage setup cleanup changed sessions"); + const deadline = await service.nextSetupCleanupAt(); + // Back off failed or still-overdue work, not a successfully read future deadline. + if (deadline === null || deadline > Date.now()) earliest = 0; + next = deadline === null ? null : Math.max(earliest, deadline); + reconciled = generation; + }, + hasPending: async () => requested !== reconciled, + nextRunAt: () => next, + }); +} diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 66f14532600..f6d50c71930 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + customImageExpiry: "custom-image-expiry", historicalCleanup: "historical-cleanup", adapterLoginCleanup: "adapter-login-cleanup", setupTokenCleanup: "setup-token-cleanup", diff --git a/server/src/services/environment-custom-images.ts b/server/src/services/environment-custom-images.ts index 324065e84d5..9ff4fd1c1df 100644 --- a/server/src/services/environment-custom-images.ts +++ b/server/src/services/environment-custom-images.ts @@ -1,7 +1,8 @@ -import { writeEnvironmentCleanup, notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; +import { notifyEnvironmentCleanup } from "./environment-cleanup-notifications.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { randomUUID } from "node:crypto"; import { withoutCleanupVerification } from "./cleanup-verification-proof.js"; -import { and, desc, eq, inArray, lte } from "drizzle-orm"; +import { and, desc, eq, inArray, lte, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { environmentCustomImageSetupSessions, @@ -290,6 +291,14 @@ export function environmentCustomImageService( const environments = environmentService(db); const secrets = secretService(db); + function writeSetup(write: (tx: Db) => PromiseLike): Promise { + return db.transaction(async tx => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.customImageExpiry); + await notifyEnvironmentCleanup(tx); + return write(tx as unknown as Db); + }); + } + async function getTemplateById(id: string): Promise { const row = await db .select() @@ -595,7 +604,7 @@ export function environmentCustomImageService( ): Promise { const status = normalizePersistedStatus(providerSession.status, fallbackStatus); const now = new Date(); - const row = await writeEnvironmentCleanup(db, tx => tx + const row = await writeSetup(tx => tx .update(environmentCustomImageSetupSessions) .set({ providerLeaseId: providerSession.providerLeaseId ?? null, @@ -620,7 +629,7 @@ export function environmentCustomImageService( metadata?: Record | null; }): Promise { const now = new Date(); - const row = await writeEnvironmentCleanup(db, tx => tx + const row = await writeSetup(tx => tx .update(environmentCustomImageSetupSessions) .set({ status: input.status, @@ -834,7 +843,7 @@ export function environmentCustomImageService( [SOURCE_ENVIRONMENT_CONFIG_FINGERPRINT_METADATA_KEY]: fingerprint, [SETUP_RPC_COMPANY_ID_METADATA_KEY]: provider.rpcCompanyId, }; - await db.insert(environmentCustomImageSetupSessions).values({ + await writeSetup(tx => tx.insert(environmentCustomImageSetupSessions).values({ id: sessionId, environmentId: input.environmentId, templateId: selectedTemplate?.id ?? null, @@ -847,7 +856,7 @@ export function environmentCustomImageService( metadata: setupMetadata, createdAt: now, updatedAt: now, - }); + })); try { const providerSession = await callProviderStart({ @@ -1020,6 +1029,7 @@ export function environmentCustomImageService( )); } await notifyEnvironmentCleanup(tx); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.customImageExpiry); await tx .update(environmentCustomImageSetupSessions) .set({ @@ -1355,6 +1365,17 @@ export function environmentCustomImageService( return template; }, + nextSetupCleanupAt: async (): Promise => { + const [row] = await db.select({ nextAt: sql`min(${environmentCustomImageSetupSessions.expiresAt})` }) + .from(environmentCustomImageSetupSessions) + .where(inArray(environmentCustomImageSetupSessions.status, [...ACTIVE_SETUP_STATUSES])); + if (!row || row.nextAt === undefined) throw new Error("Setup expiry query returned no result"); + if (row.nextAt === null) return null; + const deadline = new Date(row.nextAt).getTime(); + if (!Number.isFinite(deadline)) throw new Error("Setup expiry query returned an invalid deadline"); + return deadline; + }, + cleanupExpiredSetupSessions: async (input: { now?: Date; limit?: number; diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 91b5e3e5b6f..a0a462e6056 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -298,9 +298,16 @@ describe("automatic retained workspace recovery", () => { } finally { await coordinator.stop(); await worker.stop(); } }, 30_000); - it("fills a free export slot while another export is held and joins both at shutdown", async () => { - const held = await seed(), fast = await seed(); - const fixtures = [held, fast]; + it("fills a free export slot across the candidate cursor while another export is held and joins both at shutdown", async () => { + const orderedSeed = async (position: number) => { + const fixture = await seed(); + const id = `00000002-000${position}-4000-8000-${randomUUID().slice(-12)}` as const; + await db.update(issueRecoveryActions).set({ id }).where(eq(issueRecoveryActions.id, fixture.actionId)); + fixture.actionId = id; fixture.input.actionId = id; + return fixture; + }; + const held = await orderedSeed(2); + const fixtures = [held]; const runtimeFor = (runId: string) => fixtures.find(fixture => fixture.runId === runId)!.environmentRuntime; const runtime = { ...held.environmentRuntime, resumeWorkspaceRecoveryLease: (input: Parameters[0]) => @@ -317,15 +324,20 @@ describe("automatic retained workspace recovery", () => { try { if (pause) await gate; return await resume(input); } finally { active -= 1; } }); }; - countExport(held, true); countExport(fast); + countExport(held, true); const errors = vi.fn(); const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: errors }); const worker = registerWorkspaceRecoveryWork(db, coordinator, runtime, async () => {}); try { await expect.poll(() => held.runtime.resumeRunLease.mock.calls.length, { timeout: 20_000 }).toBe(1); + // Start the middle candidate only after the highest candidate occupies + // one slot. The next producer then sorts before that saved cursor. + const fast = await orderedSeed(1); fixtures.push(fast); countExport(fast); + await requestTaskRetry({ db, companyId: fast.companyId, issueId: fast.issueId, actorId: "board", + observedRunId: fast.runId, idempotencyKey: randomUUID() }); await expect.poll(async () => readSystemRecovery((await action(fast.actionId)).evidence)?.state, { timeout: 20_000 }).toBe("complete"); - const third = await seed(); fixtures.push(third); countExport(third); + const third = await orderedSeed(0); fixtures.push(third); countExport(third); await requestTaskRetry({ db, companyId: third.companyId, issueId: third.issueId, actorId: "board", observedRunId: third.runId, idempotencyKey: randomUUID() }); await expect.poll(async () => readSystemRecovery((await action(third.actionId)).evidence)?.state, diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index 179ddc478ad..0cf699ab239 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -2,7 +2,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio import { beginIdleTrackedWork } from "./task-admission.js"; import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js"; import { resumePreservedResultFinalization } from "./preserved-result-finalization.js"; -import { and, asc, eq, gt, inArray, ne, or, sql } from "drizzle-orm"; +import { and, asc, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm"; import { agentWakeupRequests, environmentLeases, heartbeatRuns, issues, issueRecoveryActions, workspaceOperations, type Db } from "@paperclipai/db"; import type { Environment, EnvironmentLease } from "@paperclipai/shared"; import { conflict } from "../errors.js"; @@ -313,7 +313,7 @@ export async function scheduleRecoveredWorkspaceCleanup(db: Db, run: typeof hear sql`${environmentLeases.metadata} = ${JSON.stringify(lease.metadata)}::jsonb`, sql`${environmentLeases.metadata}->>'workspaceRecoveryCleanup' is null`))); } -export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedOnly = false, obligationsOnly = false) { +export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedOnly = false, obligationsOnly = false, inFlightRunIds: string[] = []) { const cleanup = sql`(${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,resourceDisposition}' = 'destroy' and exists ( select 1 from ${environmentLeases} where ${environmentLeases.companyId} = ${heartbeatRuns.companyId} and ${environmentLeases.heartbeatRunId} = ${heartbeatRuns.id} @@ -339,6 +339,7 @@ export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedO sql`(${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' is distinct from 'recovered' or ${cleanup} or ${waitingInput})`, cursor ? gt(issueRecoveryActions.id, cursor) : undefined, + inFlightRunIds.length ? notInArray(heartbeatRuns.id, inFlightRunIds) : undefined, )).orderBy(asc(issueRecoveryActions.id)).limit(5); } @@ -351,11 +352,14 @@ export async function reconcileRetainedWorkspaces(db: Db, runtime: EnvironmentRu let pending = workInFlight.get(db); if (!pending) { pending = new Set(); workInFlight.set(db, pending); } if (pending.size >= 2) return; - let rows = await retainedWorkspaceCandidates(db, unfinishedCursors.get(db), true); - if (!rows.length) rows = await retainedWorkspaceCandidates(db, undefined, true); + // Locally running exports already own their slot. Exclude them before + // pagination so a page containing only held work cannot prevent wrapping. + const inFlightRunIds = [...pending]; + let rows = await retainedWorkspaceCandidates(db, unfinishedCursors.get(db), true, false, inFlightRunIds); + if (!rows.length) rows = await retainedWorkspaceCandidates(db, undefined, true, false, inFlightRunIds); const unfinished = rows.length > 0; - if (!rows.length) rows = await retainedWorkspaceCandidates(db, passiveSweep ? passiveSweep.cursor : cursors.get(db), false, obligationsOnly); - if (!rows.length && !passiveSweep) rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly); + if (!rows.length) rows = await retainedWorkspaceCandidates(db, passiveSweep ? passiveSweep.cursor : cursors.get(db), false, obligationsOnly, inFlightRunIds); + if (!rows.length && !passiveSweep) rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly, inFlightRunIds); const work: Promise[] = []; let hasUnvisited = false; for (const { action, run } of rows) {