mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(recovery): keep a remote restart claim's lease through slow preparation (#20540)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - After a restart, a remote run resumes on the same heartbeat run under a restart claim: a lease on its native coordinator row, 60 s for a remote claim (`REMOTE_CONTROLLER_LEASE_MS`) > - Preparation then runs before the native executor takes its own lease renewal. It includes the run-log begin and the workspace restore, which can each wait out a storage outage for minutes (#20299) > - Preparation renewed the claim only for `reattach_remote_runner`. A `bootstrap_incomplete` or resume claim kept its 60 s lease with nothing renewing it > - So a preparation wait longer than a minute let the lease lapse. The run failed at its next writer-authority check (`run_log_writer_authority_lost`) the moment storage came back > - This PR renews every remote restart claim during preparation. When the same owner's executor claims at the next attempt, it takes over that renewal > - The benefit is that a resumed remote run survives a long storage outage during preparation instead of failing as soon as storage returns ## Linked Issues or Issue Description **What happened?** In restart-chaos waves on Daytona R1 with `--storage s3 --storage-faults ST1` and K2 at +110 s, a resumed `bootstrap_incomplete` run's `heartbeat.prepare.run_output` ended `failed` right after the outage ended: - `w-20261110T101517-s11936` t01, run `9d39550a`. The claim was at 10:19:55 and the outage ended at 10:22:18.15. `run_output` failed at 10:22:18.17, after 141 s. - `w-20261110T081647-s11936` t02, run `9da10424`. The claim was at 08:21:29 and the outage ended at 08:23:30. During the outage the journal claim fails on its head `GET` before it checks authority. The first attempt after the outage reaches `claimFrom`'s `assertAuthority` (`run-log-authority.ts`), which requires a live coordinator lease. That lease expired 60 s after the claim. **Why not just widen the condition?** For a `bootstrap_incomplete` claim, the executor's own claim moves the coordinator attempt to N+1 (`nextNativeProviderAttempt`). `startNativeSessionExecutionLeaseRenewal` shares one renewal per lease owner and threw `native_session_renewal_binding_changed` when the attempt differed. That is presumably why preparation renewal was limited to reattach claims. **Related:** #20460 (live status while a run waits out a storage outage). #20514 settles a failure that this lapse used to leave silent. ## What Changed - `heartbeat/recovery.ts`: start the preparation lease renewal for every remote restart claim (`scope.remote`), not only `reattach_remote_runner`. - `native-runtime/native-execution-lease.ts`: - When the same owner's executor starts its renewal at a higher attempt, it takes over the shared renewal, rebinding the attempt and `onLost`. A lower attempt or any other binding change still throws. - A renewal that failed under the old binding is not counted as a loss. - A failed renewal is not a loss when the coordinator shows the same owner already at a higher attempt (the executor's claim committed while the renewal was in flight). The renewal deadline still ends ownership if no executor takes over. - Tests: one recovery case and three lease-renewal cases (below). ## Verification - `cd server && npx vitest run src/services/native-runtime/native-execution-lease.test.ts src/services/heartbeat/recovery.test.ts src/services/native-runtime/native-execution-lease-reacquire.test.ts src/services/native-runtime/native-runner-restart-lease-renewal.integration.test.ts src/services/native-runtime/native-runner-restart-preparation.integration.test.ts`: 143 pass. `npx tsc --noEmit`: pass. - Red/green, recovery test "keeps a remote bootstrap restart claim through preparation longer than its first lease, then hands it to the next attempt": the claim's first lease ends 4 s in. Preparation runs past it and the run-log writer authority still passes. Then the executor claims attempt 2 and its renewal keeps extending the lease with no detach. - Red with master's `recovery.ts`: the lease is never extended. - Red with only the `recovery.ts` change: `native_session_renewal_binding_changed`. - Green with both. - Red/green, lease unit tests: handover to the next attempt (and refusal of a lower one); a failure under the old binding is not a loss; a failure while the coordinator shows the same owner at a higher attempt is not a loss. All three fail on master. - Live wave: results in a comment below. ## Risks - Medium-low. - The preparation renewal now runs for remote `bootstrap_incomplete` and `resume_dead_runner` claims too. It renews the same coordinator row under the same CAS conditions reattach claims already use. - If preparation truly loses the lease (another owner takes it), its `onLost` still runs (`detachNativeSessionsForRestart`) and authority checks still fence writes. - The new non-loss case applies only when the coordinator names this exact owner at a higher attempt. The renewal deadline still bounds it. - Previous-release state: this persists nothing new. A claim saved by the previous release resumes unchanged; its preparation is simply renewed now. ## Model Used Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## 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 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
94cfb84150
commit
902f376796
4 files changed
+200
-16
No files matched your search
@@ -16,6 +16,7 @@ import type { NativeRestartRecoveryClaim, NativeRestartRecoveryDisposition } fro
|
||||
import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "../native-runtime/native-runner-ownership.js";
|
||||
import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_RESTART_RECOVERY_FAILED } from "../native-runtime/native-restart-recovery.js";
|
||||
import { startNativeSessionExecutionLeaseRenewal } from "../native-runtime/native-execution-lease.js";
|
||||
import { createRunLogWriterAuthority } from "./run-log-authority.js";
|
||||
import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, legacyShutdownWorkspaceResourceProtected, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js";
|
||||
import { adapterExecutionControls, createAdapterExecutionControl } from "../adapter-execution-control.js";
|
||||
import { currentControllerProcessStamp, currentProcessSpaceHash, currentProcessStartedAtMs } from "../controller-process-identity.js";
|
||||
@@ -1319,6 +1320,67 @@ describePostgres("heartbeat recovery module ownership", () => {
|
||||
}
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
|
||||
it("keeps a remote bootstrap restart claim through preparation longer than its first lease, then hands it to the next attempt", async () => {
|
||||
const run = await seedRun();
|
||||
const deps = callbacks(db);
|
||||
const controller = await currentNativeControllerIdentity();
|
||||
const claim: NativeRestartRecoveryClaim = { kind: "bootstrap_incomplete", runId: run.id, leaseOwner: randomUUID(),
|
||||
controllerGeneration: 3, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null };
|
||||
const scope = { companyId: run.companyId, issueId: run.issueId!, controller, remote: true };
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId, status: "running" }).where(eq(heartbeatRuns.id, run.id));
|
||||
const firstLeaseEndsAt = new Date(Date.now() + 4_000);
|
||||
await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!, phase: "observed", attempt: 1,
|
||||
controllerBootId: controller.bootId, controllerPid: controller.pid, controllerProcessStartedAt: controller.processStartedAt,
|
||||
leaseOwner: claim.leaseOwner, leaseExpiresAt: firstLeaseEndsAt });
|
||||
vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] });
|
||||
native.claims.mockImplementationOnce(async input => {
|
||||
await input.onClaim?.(claim, scope);
|
||||
return [claim];
|
||||
});
|
||||
const [admitted] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
const authority = createRunLogWriterAuthority(db, admitted, claim.leaseOwner);
|
||||
let finish!: () => void, report!: (outcome: unknown) => void;
|
||||
const outcome = new Promise<unknown>(resolve => { report = resolve; });
|
||||
const finished = new Promise<void>(resolve => { finish = resolve; });
|
||||
deps.executeRun.mockImplementationOnce(async (_runId, options) => { try {
|
||||
await vi.advanceTimersByTimeAsync(21_000);
|
||||
await vi.waitFor(async () => {
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
|
||||
expect(coordinator.leaseExpiresAt!.getTime()).toBeGreaterThan(firstLeaseEndsAt.getTime() + 10_000);
|
||||
});
|
||||
await new Promise(resolve => setTimeout(resolve, Math.max(0, firstLeaseEndsAt.getTime() - Date.now() + 1_000)));
|
||||
await authority();
|
||||
await db.update(nativeRunFinalizations).set({ attempt: 2 }).where(eq(nativeRunFinalizations.runId, run.id));
|
||||
const execution = startNativeSessionExecutionLeaseRenewal({ db, ...scope, runId: run.id,
|
||||
leaseOwner: claim.leaseOwner, attempt: 2, onLost: () => native.detach() });
|
||||
try {
|
||||
await options?.onNativeExecutionLeaseAcquired?.();
|
||||
const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
|
||||
await vi.advanceTimersByTimeAsync(21_000);
|
||||
await vi.waitFor(async () => {
|
||||
const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
|
||||
expect(after.leaseExpiresAt!.getTime()).toBeGreaterThan(before.leaseExpiresAt!.getTime());
|
||||
});
|
||||
report("handed off");
|
||||
await finished;
|
||||
} finally { await execution.stop(); }
|
||||
} catch (error) { report(error); } });
|
||||
try {
|
||||
native.detach.mockClear();
|
||||
await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart(new Date(), [run.id]);
|
||||
expect(await outcome).toBe("handed off");
|
||||
expect(deps.executeRun).toHaveBeenCalledWith(run.id, expect.objectContaining({ nativeLeaseOwner: claim.leaseOwner,
|
||||
onNativeExecutionLeaseAcquired: expect.any(Function) }));
|
||||
await vi.waitFor(() => expect(native.detach).not.toHaveBeenCalled());
|
||||
} finally {
|
||||
finish();
|
||||
await Promise.all([...deps.activeRunExecutionPromises]);
|
||||
vi.useRealTimers();
|
||||
await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id));
|
||||
await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
}
|
||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||
|
||||
it.each(["bootstrap_incomplete", "blocked"] as const)("retries durable remote takeover and preserves its %s disposition", async kind => {
|
||||
const run = await seedRun();
|
||||
const deps = callbacks(db);
|
||||
|
||||
@@ -999,7 +999,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD
|
||||
try {
|
||||
const dispositions = await claimNativeRestartRecoveries({
|
||||
onClaim: (claim, scope) => {
|
||||
if (!scope.remote || claim.kind !== "reattach_remote_runner") return;
|
||||
if (!scope.remote) return;
|
||||
// Start before the rest of the batch or its event writes can delay dispatch.
|
||||
preparationRenewals.set(claim.runId, startNativeSessionExecutionLeaseRenewal({
|
||||
db, ...scope, runId: claim.runId, leaseOwner: claim.leaseOwner,
|
||||
|
||||
@@ -4,10 +4,13 @@ import { startNativeSessionExecutionLeaseRenewal } from "./native-execution-leas
|
||||
|
||||
function fixture() {
|
||||
const returning = vi.fn(async () => [{ runId: "run" }]);
|
||||
const update = vi.fn(() => ({ set: () => ({ where: () => ({ returning }) }) }));
|
||||
const transaction = vi.fn(async (run: (tx: unknown) => Promise<unknown>) => run({ update, execute: vi.fn() }));
|
||||
const coordinator = vi.fn(async (): Promise<Array<{ attempt: number; leaseOwner: string }>> => []);
|
||||
const renewed = vi.fn((_values: unknown) => undefined);
|
||||
const update = vi.fn(() => ({ set: (values: unknown) => { renewed(values); return { where: (condition: unknown) => { renewed(condition); return { returning }; } }; } }));
|
||||
const select = () => ({ from: () => ({ where: () => ({ limit: coordinator }) }) });
|
||||
const transaction = vi.fn(async (run: (tx: unknown) => Promise<unknown>) => run({ update, select, execute: vi.fn() }));
|
||||
const db = { update, transaction };
|
||||
return { update, returning, transaction, input: { db: db as unknown as Db, runId: "run", companyId: "company", issueId: "issue",
|
||||
return { update, returning, transaction, coordinator, input: { db: db as unknown as Db, runId: "run", companyId: "company", issueId: "issue",
|
||||
leaseOwner: "owner", attempt: 1, controller: { bootId: "boot", pid: 1, processStartedAt: new Date() }, remote: true,
|
||||
onLost: vi.fn(async () => undefined) } };
|
||||
}
|
||||
@@ -88,6 +91,100 @@ describe("native execution renewal ownership", () => {
|
||||
expect(returning).toHaveBeenCalledTimes(1);
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
it("hands preparation renewal to the same owner's executor at the next attempt", async () => {
|
||||
vi.useFakeTimers();
|
||||
const { input, returning } = fixture();
|
||||
const preparation = startNativeSessionExecutionLeaseRenewal(input);
|
||||
expect(() => startNativeSessionExecutionLeaseRenewal({ ...input, attempt: 0 })).toThrow("native_session_renewal_binding_changed");
|
||||
const executorLost = vi.fn(async () => undefined);
|
||||
const execution = startNativeSessionExecutionLeaseRenewal({ ...input, attempt: 2, onLost: executorLost });
|
||||
await preparation.stop();
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
expect(returning).toHaveBeenCalledTimes(1);
|
||||
returning.mockResolvedValueOnce([]);
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
expect(executorLost).toHaveBeenCalledTimes(1);
|
||||
expect(input.onLost).not.toHaveBeenCalled();
|
||||
await expect(execution.stop()).rejects.toThrow("native_session_lease_lost");
|
||||
});
|
||||
it("gives the executor that takes over at the next attempt its own later deadline", async () => {
|
||||
vi.useFakeTimers();
|
||||
const { input, returning } = fixture();
|
||||
let complete!: (value: { runId: string }[]) => void;
|
||||
const preparation = startNativeSessionExecutionLeaseRenewal({ ...input, confirmedDeadline: performance.now() + 25_000 });
|
||||
returning.mockImplementationOnce(() => new Promise(resolve => { complete = resolve; }));
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
const executorLost = vi.fn(async () => undefined);
|
||||
const execution = startNativeSessionExecutionLeaseRenewal({ ...input, attempt: 2, onLost: executorLost,
|
||||
confirmedDeadline: performance.now() + 60_000 });
|
||||
expect(execution.remainingMs()).toBe(60_000);
|
||||
const released = preparation.stop().catch(() => undefined);
|
||||
await vi.advanceTimersByTimeAsync(30_000);
|
||||
expect(executorLost).not.toHaveBeenCalled();
|
||||
complete([{ runId: "run" }]);
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(execution.remainingMs()).toBeGreaterThanOrEqual(30_000);
|
||||
await released;
|
||||
await execution.stop();
|
||||
});
|
||||
it("reacquires a lapsed local lease with the generation of the attempt that took over", async () => {
|
||||
vi.useFakeTimers();
|
||||
const { input, update, returning } = fixture();
|
||||
const conditions: unknown[] = [];
|
||||
update.mockImplementation(() => ({ set: () => ({ where: (condition: unknown) => { conditions.push(condition); return { returning }; } }) }));
|
||||
const local = { ...input, remote: false, controllerGeneration: 3 };
|
||||
const preparation = startNativeSessionExecutionLeaseRenewal(local);
|
||||
const execution = startNativeSessionExecutionLeaseRenewal({ ...local, attempt: 2, controllerGeneration: 4 });
|
||||
await preparation.stop();
|
||||
await vi.advanceTimersByTimeAsync(5 * 60_000);
|
||||
const values = (root: unknown) => {
|
||||
const found = new Set<unknown>(), seen = new WeakSet<object>();
|
||||
const walk = (node: unknown) => {
|
||||
if (node === null || typeof node !== "object") { found.add(node); return; }
|
||||
if (seen.has(node)) return; seen.add(node);
|
||||
for (const child of Object.values(node)) walk(child);
|
||||
};
|
||||
walk(root);
|
||||
return found;
|
||||
};
|
||||
expect(conditions.length).toBeGreaterThan(0);
|
||||
expect(values(conditions.at(-1)).has(4)).toBe(true);
|
||||
expect(values(conditions.at(-1)).has(3)).toBe(false);
|
||||
await execution.stop();
|
||||
});
|
||||
it("does not count a renewal that failed because the owner's executor advanced the attempt", async () => {
|
||||
vi.useFakeTimers();
|
||||
const { input, returning } = fixture();
|
||||
let complete!: (value: { runId: string }[]) => void;
|
||||
returning.mockImplementationOnce(() => new Promise(resolve => { complete = resolve; }));
|
||||
const preparation = startNativeSessionExecutionLeaseRenewal(input);
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
expect(returning).toHaveBeenCalledTimes(1);
|
||||
const execution = startNativeSessionExecutionLeaseRenewal({ ...input, attempt: 2 });
|
||||
complete([]);
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
expect(returning).toHaveBeenCalledTimes(2);
|
||||
expect(input.onLost).not.toHaveBeenCalled();
|
||||
await preparation.stop();
|
||||
await execution.stop();
|
||||
expect(vi.getTimerCount()).toBe(0);
|
||||
});
|
||||
it("keeps renewing when the coordinator shows its own owner already claimed the next attempt", async () => {
|
||||
vi.useFakeTimers();
|
||||
const { input, returning, coordinator } = fixture();
|
||||
coordinator.mockResolvedValue([{ attempt: 2, leaseOwner: "owner" }]);
|
||||
returning.mockResolvedValueOnce([]);
|
||||
const preparation = startNativeSessionExecutionLeaseRenewal(input);
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
expect(coordinator).toHaveBeenCalledTimes(1);
|
||||
expect(input.onLost).not.toHaveBeenCalled();
|
||||
const execution = startNativeSessionExecutionLeaseRenewal({ ...input, attempt: 2 });
|
||||
await preparation.stop();
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
expect(returning).toHaveBeenCalledTimes(2);
|
||||
expect(input.onLost).not.toHaveBeenCalled();
|
||||
await execution.stop();
|
||||
});
|
||||
it("does not queue SQL behind a slow renewal and fences a lost owner", async () => {
|
||||
vi.useFakeTimers();
|
||||
const { input, update, returning } = fixture();
|
||||
|
||||
@@ -253,20 +253,37 @@ const renewals = new WeakMap<Db, Map<string, {
|
||||
input: RenewalInput; references: number; timer: ReturnType<typeof setInterval>;
|
||||
pending: Promise<void>; busy: boolean; lost: Error | null;
|
||||
deadline: ReturnType<typeof setTimeout>; loss: Promise<unknown>;
|
||||
confirmedDeadline: number;
|
||||
confirmedDeadline: number; adopt(confirmedDeadline: number): void;
|
||||
}>>();
|
||||
|
||||
async function attemptAdvancedUnderOwner(binding: RenewalInput): Promise<boolean> {
|
||||
const [row] = await livenessDb(binding.db).transaction(async tx => {
|
||||
await tx.execute(sql`select set_config('statement_timeout', ${`${RENEWAL_STATEMENT_TIMEOUT_MS}ms`}, true)`);
|
||||
return tx.select({ attempt: nativeRunFinalizations.attempt, leaseOwner: nativeRunFinalizations.leaseOwner })
|
||||
.from(nativeRunFinalizations)
|
||||
.where(and(eq(nativeRunFinalizations.runId, binding.runId), eq(nativeRunFinalizations.companyId, binding.companyId)))
|
||||
.limit(1);
|
||||
});
|
||||
return row?.leaseOwner === binding.leaseOwner && row.attempt > binding.attempt;
|
||||
}
|
||||
|
||||
export function startNativeSessionExecutionLeaseRenewal(input: RenewalInput): NativeExecutionLeaseRenewal {
|
||||
let owners = renewals.get(input.db);
|
||||
if (!owners) { owners = new Map(); renewals.set(input.db, owners); }
|
||||
let state = owners.get(input.leaseOwner);
|
||||
if (state) {
|
||||
if (state.input.runId !== input.runId || state.input.companyId !== input.companyId
|
||||
|| state.input.issueId !== input.issueId || state.input.attempt !== input.attempt
|
||||
|| state.input.issueId !== input.issueId || state.input.attempt > input.attempt
|
||||
|| state.input.controller.bootId !== input.controller.bootId || state.input.remote !== input.remote) {
|
||||
throw new Error("native_session_renewal_binding_changed");
|
||||
}
|
||||
if (state.lost) throw state.lost;
|
||||
if (input.attempt > state.input.attempt) {
|
||||
state.input = { ...state.input, attempt: input.attempt, onLost: input.onLost, controller: input.controller,
|
||||
controllerGeneration: input.controllerGeneration ?? state.input.controllerGeneration };
|
||||
state.adopt(input.confirmedDeadline
|
||||
?? performance.now() + (input.remote ? REMOTE_CONTROLLER_LEASE_MS : NATIVE_SESSION_EXECUTION_LEASE_TTL_MS));
|
||||
}
|
||||
state.references += 1;
|
||||
} else {
|
||||
const ttlMs = input.remote ? REMOTE_CONTROLLER_LEASE_MS : NATIVE_SESSION_EXECUTION_LEASE_TTL_MS;
|
||||
@@ -275,7 +292,7 @@ export function startNativeSessionExecutionLeaseRenewal(input: RenewalInput): Na
|
||||
state!.lost = error instanceof Error ? error : new Error("native_session_lease_lost");
|
||||
clearInterval(state!.timer);
|
||||
clearTimeout(state!.deadline);
|
||||
state!.loss = Promise.resolve().then(input.onLost).catch(() => undefined);
|
||||
state!.loss = Promise.resolve().then(state!.input.onLost).catch(() => undefined);
|
||||
};
|
||||
const deadline = (delay: number) => {
|
||||
const timer = setTimeout(() => lose(new Error("native_session_lease_renewal_deadline")), Math.max(0, delay));
|
||||
@@ -285,18 +302,18 @@ export function startNativeSessionExecutionLeaseRenewal(input: RenewalInput): Na
|
||||
const renew = () => {
|
||||
if (state!.lost || state!.busy) return;
|
||||
state!.busy = true;
|
||||
state!.pending = renewNativeSessionExecutionLeaseDeadline({ ...input, leaseTtlMs: ttlMs,
|
||||
reacquire: !input.remote && input.controllerGeneration !== undefined
|
||||
? { controllerGeneration: input.controllerGeneration } : undefined })
|
||||
const binding = state!.input;
|
||||
state!.pending = renewNativeSessionExecutionLeaseDeadline({ ...binding, leaseTtlMs: ttlMs,
|
||||
reacquire: !binding.remote && binding.controllerGeneration !== undefined
|
||||
? { controllerGeneration: binding.controllerGeneration } : undefined })
|
||||
.then(confirmedDeadline => {
|
||||
// A late database response cannot restore local authority after loss
|
||||
// or recreate a timer after the final caller has stopped.
|
||||
if (state!.lost || state!.references === 0) return;
|
||||
clearTimeout(state!.deadline);
|
||||
state!.confirmedDeadline = confirmedDeadline;
|
||||
state!.deadline = deadline(confirmedDeadline - performance.now());
|
||||
if (state!.references === 0) return;
|
||||
state!.adopt(confirmedDeadline);
|
||||
})
|
||||
.catch((error: unknown) => {
|
||||
.catch(async (error: unknown) => {
|
||||
if (state!.input !== binding) return;
|
||||
// A cancelled statement can retry at the next tick, but does not
|
||||
// extend the ownership deadline. Never queue overlapping attempts.
|
||||
for (let cause = error, depth = 0; cause && depth < 4; cause = (cause as { cause?: unknown }).cause, depth++) {
|
||||
@@ -305,14 +322,22 @@ export function startNativeSessionExecutionLeaseRenewal(input: RenewalInput): Na
|
||||
const code = (cause as { code?: unknown }).code;
|
||||
if (code === "57014" || code === "40P01") return;
|
||||
}
|
||||
const advanced = await attemptAdvancedUnderOwner(binding).catch(() => null);
|
||||
if (advanced !== false || state!.input !== binding) return;
|
||||
lose(error);
|
||||
}).finally(() => { state!.busy = false; });
|
||||
};
|
||||
const timer = setInterval(renew, input.remote ? REMOTE_CONTROLLER_LEASE_MS / 3 : NATIVE_SESSION_EXECUTION_LEASE_RENEW_INTERVAL_MS);
|
||||
timer.unref?.();
|
||||
const confirmedDeadline = input.confirmedDeadline ?? performance.now() + ttlMs;
|
||||
const adopt = (confirmed: number) => {
|
||||
if (state!.lost || confirmed <= state!.confirmedDeadline) return;
|
||||
clearTimeout(state!.deadline);
|
||||
state!.confirmedDeadline = confirmed;
|
||||
state!.deadline = deadline(confirmed - performance.now());
|
||||
};
|
||||
state = { input, references: 1, timer, pending: Promise.resolve(), busy: false, lost: null,
|
||||
deadline: deadline(confirmedDeadline - performance.now()), confirmedDeadline, loss: Promise.resolve() };
|
||||
deadline: deadline(confirmedDeadline - performance.now()), confirmedDeadline, loss: Promise.resolve(), adopt };
|
||||
owners.set(input.leaseOwner, state);
|
||||
}
|
||||
const owned = state;
|
||||
|
||||
Reference in new issue
Block a user