mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 12:04:41 +01:00
fix(server): schedule custom-image setup expiry from committed changes (#17443)
## 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 <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
50b932a142
commit
5982246771
18 files changed
+569
-61
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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,
|
||||
|
||||
@@ -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.
|
||||
@@ -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.
|
||||
+1
-1
@@ -2656,4 +2656,4 @@
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -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();
|
||||
}
|
||||
|
||||
|
||||
@@ -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<typeof createDb>;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator>;
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>, db: ReturnType<typeof createDb>;
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined, db: ReturnType<typeof createDb>;
|
||||
let companyId: string, environmentId: string, draining: boolean, failing: boolean;
|
||||
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let worker: ReturnType<typeof registerEnvironmentCleanupWork> | undefined;
|
||||
let releases: ReturnType<typeof vi.fn>, inspections: ReturnType<typeof vi.fn>, reads: ReturnType<typeof vi.spyOn>;
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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<typeof createDb>;
|
||||
|
||||
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<typeof createDeliveryWorkCoordinator> | undefined;
|
||||
let service: ReturnType<typeof environmentCustomImageService>;
|
||||
let provider: ReturnType<typeof createWorkerManager>;
|
||||
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<void>(resolve => { release = resolve; });
|
||||
const staged = new Promise<void>(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<void>(resolve => { release = resolve; });
|
||||
const entered = new Promise<void>(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<typeof createDb>;
|
||||
|
||||
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<typeof createDb>;
|
||||
|
||||
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?.();
|
||||
});
|
||||
|
||||
|
||||
@@ -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([]);
|
||||
|
||||
@@ -129,6 +129,7 @@ const {
|
||||
};
|
||||
const issueThreadInteractionServiceFactoryMock = vi.fn(() => issueThreadInteractionServiceMock);
|
||||
const environmentCustomImagesServiceMock = {
|
||||
nextSetupCleanupAt: vi.fn(async (): Promise<number | null> => 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 {
|
||||
|
||||
+4
-18
@@ -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()
|
||||
|
||||
@@ -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<typeof createDeliveryWorkCoordinator>,
|
||||
service: Pick<ReturnType<typeof environmentCustomImageService>, "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,
|
||||
});
|
||||
}
|
||||
@@ -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",
|
||||
|
||||
@@ -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<T>(write: (tx: Db) => PromiseLike<T>): Promise<T> {
|
||||
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<EnvironmentCustomImageTemplate | null> {
|
||||
const row = await db
|
||||
.select()
|
||||
@@ -595,7 +604,7 @@ export function environmentCustomImageService(
|
||||
): Promise<EnvironmentCustomImageSetupSession> {
|
||||
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<string, unknown> | null;
|
||||
}): Promise<EnvironmentCustomImageSetupSession> {
|
||||
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<number | null> => {
|
||||
const [row] = await db.select({ nextAt: sql<Date | string | null>`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;
|
||||
|
||||
@@ -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<typeof held.environmentRuntime.resumeWorkspaceRecoveryLease>[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,
|
||||
|
||||
@@ -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<void>[] = [];
|
||||
let hasUnvisited = false;
|
||||
for (const { action, run } of rows) {
|
||||
|
||||
Reference in new issue
Block a user