mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix(runner): answer lease renewal outside the PRP frame chain and name lease loss (#19419)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native runs in a managed sandbox talk to the server over one authenticated PRP connection. The runner (`paperclip-runnerd`) renews its connection lease on that connection. > - The server serializes every inbound PRP frame through one promise chain per connection. Before this change, `lease_renew` waited in that chain behind earlier events. An event whose business commit stalls (for example the commit of a tool event) therefore also stalls the lease reply. The runner then stops polling its provider (ACK backpressure) and exits with "active connection lease expired" when the lease runs out. > - When that happens, the server sees only a socket close. The run log and the terminal error say "transport loss", not "lease expired". > - A follow-up from #19407 asked for renewal to be handled outside the chain and for lease loss to be named. > - This pull request answers `lease_renew` (and `pong`) as soon as the frame is decrypted. It keeps in-order decryption and the order of state-changing frames. It also records why a lease was lost and reports it as `runner_connection_lease_expired`. > - The benefit is that a slow event commit can no longer starve lease renewal, and a lease loss is visible in the run log and error code. ## Linked Issues or Issue Description Refs #19407 **What happened?** On staging, native runs with ACPX providers (Grok, Kimi, and Pi with OpenRouter/DeepSeek) ran a long foreground shell command that printed nothing (a 15–20 minute `sleep 60` loop). Each run went silent soon after `tool.execution.started`. The command finished in the sandbox, but the run never got the completion and stayed `running` until someone cancelled it. It happened in 3 of 3 attempts. The same task on Codex kept reporting activity and succeeded. **What did you expect to happen?** A quiet tool call keeps its runner connection and its lease. If the connection or lease is lost, the run log and the error code say why. **Steps to reproduce** 1. Assign a native ACPX agent a task that runs a foreground shell loop with no output for longer than 2 minutes. 2. Observe that the run goes silent and does not finish after the command ends. **Root cause analysis** - The ACPX path emits no PRP frames during a silent tool call. The runner has no timer, and the agent sends no `tool_call_update`. Codex keeps the connection busy, because its model polls the running command. Silence is the only adapter-specific factor that I found. I found no ACPX code path in runnerd that blocks the main loop: polls use a 1 ms timeout, and synchronous sidecar requests run only for server commands. - The staging build (`de1143448b`) did not include #19407. That build had no server pings. With a 60-minute connection lease (the default for `runner-prp-coordinator` and `DurablePrpCodexTransport`), the runner renews only every 30 minutes. A fully silent connection that goes through the sandbox preview proxy therefore carried no traffic at all. A route that the proxy drops without a FIN then becomes half-open forever. #19407 added 20-second pings and a 90-second termination, which closes that window. - The lease hypothesis from #19407 is real but secondary. The 60-second lease applies only when the TTL is not set. Renewal waited behind any slow event commit, and the server could not report lease loss. This pull request fixes both. ## What Changed - `DurablePrpControlPlane.attachWireConnection` decrypts each secure frame when it arrives, in counter order. Frames that arrive before the channel is secure, or behind such a frame, still decrypt in the chain. - `lease_renew` and `pong` are handled at once, outside the per-connection chain. Admission against persisted authority is unchanged. Events, command results, and handoff frames keep their serialized order. - `#renewLease` persists the renewed expiry with `save()`, so an event write that is already staged is absorbed into the same write. Before, the renewal used `commit()`, which would invalidate that staged event and close a healthy connection when renewal runs concurrently. - New `connectionLeaseFailure` getter (exported type `DurablePrpConnectionLeaseFailure`). It records `connection_lease_expired` when an authenticated connection is refused or closes with an expired lease. A new authenticated connection clears it. A credential revoked by a warm handoff is not reported, because a replacement runner owns the run. - `connectRunnerPrpIngress` reads that reason after a close. It logs `state=reconnecting failure=runner_connection_lease_expired` in the run log. If reconnect grace runs out, it fails with that code instead of `runner_ingress_unavailable` or `runner_ingress_auth_failed`, and keeps the reconnect error as the cause. - `packages/paperclip-runner/docs/durable-recovery.md` documents the new behavior. ## Verification - `pnpm --filter @paperclipai/paperclip-runner exec vitest run src/control-plane` (220 passed). - New: "answers lease renewal while an earlier event commit is still in flight". It was red before the fix (no `lease_renewed` while `onCommittedEvent` waits), and it is green now. It also checks that the event later commits and ACKs, and that the renewed expiry survives. - New: "keeps a staged tool-progress write when a lease renewal lands mid-write". It holds a file-backed `tool.execution.progressed` write at `fsync`. It is red with the old `commit()` renewal (the event is dropped and the connection closes), and green with `save()`. - New: "reports an expired connection lease instead of closing silently". It was red before the fix and is green now. - `pnpm --filter @paperclipai/server exec vitest run src/realtime/runner-prp-outbound.test.ts src/realtime/runner-prp-ws.test.ts src/services/native-runtime/runner-prp-coordinator.test.ts` (all passed). - New: "names an expired runner connection lease in the run log and terminal error when reconnect is refused/unauthenticated". It was red before the fix and is green now. - `src/live/runnerd-codex-transport.test.ts` has the same result with and without this change with a locally built `paperclip-runnerd`: the same 48 tests fail locally on both, for environment reasons, and the other 207 pass. ## Risks - Lease renewal now runs concurrently with an in-flight event handler. The handler builds its state candidate at commit time, and `save()` absorbs a staged event write. A new regression test covers the file-backed race. External checkpoint persistence commits events synchronously, and the renewal reply still waits for the checkpoint flush. - Renewal still stays silent while a warm handoff or a terminal lifecycle command is pending. The runner treats any other control frame inside those fences as fatal, so this pull request does not change that behavior. - No protocol, schema, or contract-digest change. The runner is unchanged. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude Code with tool use (shell, file edits, test runs) and a read-only exploration subagent on the same model. ## 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 - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 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
4d49a31286
commit
39096a55a4
6 files changed
+399
-24
No files matched your search
@@ -41,6 +41,19 @@ lease without replacing its process, provider, thread, turn, token, or epoch.
|
||||
Retries of the same observed expiry replay the persisted extension. Renewal
|
||||
never admits an expired, revoked, or differently bound credential.
|
||||
|
||||
The controller answers `lease_renew` (and `pong`) as soon as the frame is
|
||||
decrypted, outside the per-connection chain that serializes events, command
|
||||
results, and handoff frames. A slow business commit for an earlier event, such
|
||||
as a tool event during a long silent tool call, therefore cannot starve
|
||||
renewal. Frames still decrypt in arrival order, and the renewed expiry is
|
||||
persisted (absorbing any event write already staged) before the reply is sent.
|
||||
When an authenticated connection closes or is refused because its lease has
|
||||
expired, the controller records that reason (`connectionLeaseFailure`).
|
||||
Provider ingress reports it in the run log and, if reconnect grace runs out,
|
||||
as the `runner_connection_lease_expired` failure code instead of a generic
|
||||
transport loss. A credential revoked by a warm handoff is not reported, because
|
||||
a replacement runner owns the run.
|
||||
|
||||
If a reply is lost, reconnect authentication may reconcile a later expiry only
|
||||
when this runner has an outstanding renewal on that same credential. Warm
|
||||
handoff receipts continue to bind exact expiry and renewal pauses during their
|
||||
|
||||
@@ -1399,6 +1399,153 @@ it("renews one authenticated connection for three weeks without replacing its au
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
async function receiveSecureWithin(client: AuthenticatedClient, ms: number): Promise<Record<string, unknown> | "timeout" | null> {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
return await Promise.race([
|
||||
receiveSecure(client),
|
||||
new Promise<"timeout">(resolveTimeout => { timer = setTimeout(() => resolveTimeout("timeout"), ms); }),
|
||||
]);
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
function credentialEvent(sourceSeq: number): Record<string, unknown> {
|
||||
const frame = semanticInputEvent(sourceSeq);
|
||||
const event = frame.payload as Record<string, unknown>;
|
||||
event.sourceEventId = `slow-commit-${sourceSeq}`;
|
||||
event.eventType = "auth.credentials.updated";
|
||||
event.payload = { schema: "paperclip.auth.credentials.updated.v1", provider: "codex",
|
||||
bindingId: "harness-auth", revision: sourceSeq, detection: "credential_store" };
|
||||
return frame;
|
||||
}
|
||||
|
||||
it("answers lease renewal while an earlier event commit is still in flight", async () => {
|
||||
const root = mkdtempSync(resolve(tmpdir(), "runner-lease-renewal-slow-commit-"));
|
||||
let release!: () => void;
|
||||
const barrier = new Promise<void>(resolveCommit => { release = resolveCommit; });
|
||||
const commits = vi.fn(async () => barrier);
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: root, identity,
|
||||
expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: commits });
|
||||
try {
|
||||
await core.start();
|
||||
const client = (await authenticate(core, core.issueBootstrapTicket()))!;
|
||||
const expiry = Number((client.welcome.payload as Record<string, unknown>).connectionLeaseExpiresAtUnixMs);
|
||||
// A tool event whose business commit stalls must not starve the lease.
|
||||
sendSecure(client, credentialEvent(1));
|
||||
await vi.waitFor(() => expect(commits).toHaveBeenCalledTimes(1));
|
||||
sendSecure(client, renewalRequest(client, expiry));
|
||||
const reply = await receiveSecureWithin(client, 2_000);
|
||||
expect(reply).not.toBe("timeout");
|
||||
expect(reply).toMatchObject({ kind: "lease_renewed", payload: { previousExpiresAtUnixMs: expiry } });
|
||||
const renewed = Number(((reply as Record<string, unknown>).payload as Record<string, unknown>).connectionLeaseExpiresAtUnixMs);
|
||||
expect(renewed).toBeGreaterThanOrEqual(expiry);
|
||||
expect(core.store.state.ackedSourceSeq).toBe(0);
|
||||
// Ordering for state-changing frames is unchanged: the event still
|
||||
// commits, ACKs, and the renewed lease survives that later commit.
|
||||
release();
|
||||
expect(await receiveSecure(client)).toMatchObject({ kind: "ack", payload: { ackedSourceSeq: 1 } });
|
||||
sendSecure(client, credentialEvent(2));
|
||||
expect(await receiveSecure(client)).toMatchObject({ kind: "ack", payload: { ackedSourceSeq: 2 } });
|
||||
const saved = JSON.parse(readFileSync(core.store.path, "utf8"));
|
||||
expect(saved.ackedSourceSeq).toBe(2);
|
||||
expect(saved.leases[Object.keys(saved.leases)[0]!].expiresAtUnixMs).toBe(renewed);
|
||||
client.socket.destroy();
|
||||
} finally {
|
||||
release();
|
||||
await core.stop();
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps a staged tool-progress write when a lease renewal lands mid-write", async () => {
|
||||
const root = mkdtempSync(resolve(tmpdir(), "runner-lease-renewal-staged-write-"));
|
||||
let release!: () => void;
|
||||
const barrier = new Promise<void>(resolveWrite => { release = resolveWrite; });
|
||||
let staged = false;
|
||||
let hold = true;
|
||||
const originalOpen = fsPromises.open;
|
||||
const open = vi.spyOn(fsPromises, "open").mockImplementation(async (...args) => {
|
||||
const file = await originalOpen(...args);
|
||||
if (hold && String(args[0]).includes(".control-plane-state.json.")) {
|
||||
hold = false;
|
||||
const sync = file.sync.bind(file);
|
||||
file.sync = async () => {
|
||||
staged = true;
|
||||
await barrier;
|
||||
await sync();
|
||||
};
|
||||
}
|
||||
return file;
|
||||
});
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: root, identity,
|
||||
expectedRunnerVersion, expectedRunnerDigest });
|
||||
try {
|
||||
await core.start();
|
||||
const client = (await authenticate(core, core.issueBootstrapTicket()))!;
|
||||
const expiry = Number((client.welcome.payload as Record<string, unknown>).connectionLeaseExpiresAtUnixMs);
|
||||
const event = semanticInputEvent();
|
||||
const payload = event.payload as Record<string, unknown>;
|
||||
payload.eventType = "tool.execution.progressed";
|
||||
payload.priority = 2;
|
||||
payload.payload = { schema: "paperclip.tool.execution.v1", executionId: "long-shell",
|
||||
transport: "process", operation: "execute", status: "running", output: "",
|
||||
outputBytes: 0, outputTruncated: false, outputDigest: null };
|
||||
expect(validatePrpEvent(payload)).toMatchObject({ ok: true });
|
||||
sendSecure(client, event);
|
||||
await vi.waitFor(() => expect(staged).toBe(true));
|
||||
sendSecure(client, renewalRequest(client, expiry));
|
||||
const reply = await receiveSecureWithin(client, 2_000);
|
||||
expect(reply).toMatchObject({ kind: "lease_renewed" });
|
||||
release();
|
||||
// The renewal absorbed the admitted event instead of invalidating it.
|
||||
expect(await receiveSecure(client)).toMatchObject({ kind: "ack", payload: { ackedSourceSeq: 1 } });
|
||||
const saved = JSON.parse(readFileSync(core.store.path, "utf8"));
|
||||
expect(saved.ackedSourceSeq).toBe(1);
|
||||
expect(saved.leases[Object.keys(saved.leases)[0]!].expiresAtUnixMs).toBe(
|
||||
((reply as Record<string, unknown>).payload as Record<string, unknown>).connectionLeaseExpiresAtUnixMs);
|
||||
client.socket.destroy();
|
||||
} finally {
|
||||
release();
|
||||
open.mockRestore();
|
||||
await core.stop();
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("reports an expired connection lease instead of closing silently", async () => {
|
||||
const root = mkdtempSync(resolve(tmpdir(), "runner-lease-expired-"));
|
||||
let now = Date.now();
|
||||
const clock = vi.spyOn(Date, "now").mockImplementation(() => now);
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: root, identity,
|
||||
expectedRunnerVersion, expectedRunnerDigest, connectionLeaseTtlMs: 60_000 });
|
||||
try {
|
||||
await core.start();
|
||||
const client = (await authenticate(core, core.issueBootstrapTicket()))!;
|
||||
const expiry = Number((client.welcome.payload as Record<string, unknown>).connectionLeaseExpiresAtUnixMs);
|
||||
expect(core.connectionLeaseFailure).toBeNull();
|
||||
now = expiry + 1;
|
||||
sendSecure(client, renewalRequest(client, expiry));
|
||||
expect(await receiveSecureWithin(client, 2_000)).toBeNull();
|
||||
expect(core.connectionLeaseFailure).toMatchObject({
|
||||
code: "connection_lease_expired",
|
||||
leaseId: client.welcome.connectionLeaseId,
|
||||
expiresAtUnixMs: expiry,
|
||||
});
|
||||
// A fresh authenticated connection clears the stale reason.
|
||||
now = Date.now();
|
||||
const replacement = (await authenticate(core, core.issueBootstrapTicket()))!;
|
||||
expect(core.connectionLeaseFailure).toBeNull();
|
||||
replacement.socket.destroy();
|
||||
client.socket.destroy();
|
||||
} finally {
|
||||
clock.mockRestore();
|
||||
await core.stop();
|
||||
rmSync(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("reports how long the authenticated Runner has been disconnected", async () => {
|
||||
const root = mkdtempSync(resolve(tmpdir(), "runner-disconnect-age-"));
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest });
|
||||
|
||||
@@ -1774,6 +1774,18 @@ class DurableCoreStore {
|
||||
}
|
||||
}
|
||||
|
||||
/** Why an authenticated runner connection lost its lease authority. */
|
||||
export interface DurablePrpConnectionLeaseFailure {
|
||||
readonly code: "connection_lease_expired";
|
||||
readonly leaseId: string;
|
||||
readonly expiresAtUnixMs: number;
|
||||
readonly observedAtUnixMs: number;
|
||||
}
|
||||
|
||||
type SerializedFrame =
|
||||
| { readonly raw: unknown }
|
||||
| { readonly decrypted: Record<string, unknown> };
|
||||
|
||||
/** Reason supplied when a transport-neutral PRP peer closes. */
|
||||
export interface TransportCloseReason {
|
||||
readonly code?: number;
|
||||
@@ -2004,6 +2016,7 @@ export class DurablePrpControlPlane {
|
||||
// next authentication. Diagnostic only: it never gates authority.
|
||||
#runnerDisconnectedAt: number | null = null;
|
||||
#connectionProcessing = new Map<AuthorityConnection, Promise<void>>();
|
||||
#connectionLeaseFailure: DurablePrpConnectionLeaseFailure | null = null;
|
||||
#pendingSemanticCalls = new Set<string>();
|
||||
#semanticCallbacksRetired = false;
|
||||
#semanticResultPersistenceFailed = false;
|
||||
@@ -2912,11 +2925,40 @@ export class DurablePrpControlPlane {
|
||||
attachWireConnection(wire: PrpWireConnection): PrpWireAttachment {
|
||||
let connection!: AuthorityConnection;
|
||||
let processing = Promise.resolve();
|
||||
// Frames received before the secure channel existed (or behind such a
|
||||
// frame) are decrypted in order by the serialized chain. Once none are
|
||||
// pending, frames are decrypted on arrival so that lease renewal can be
|
||||
// answered without waiting for an earlier event's business commit.
|
||||
let undecryptedFrames = 0;
|
||||
connection = new AuthorityConnection({
|
||||
wire,
|
||||
onJson: (value) => {
|
||||
let frame: SerializedFrame;
|
||||
if (connection.secureChannel !== null && undecryptedFrames === 0) {
|
||||
const decrypted = this.#decryptOnArrival(connection, value);
|
||||
if (decrypted === null) return;
|
||||
if (decrypted.kind === "lease_renew" || decrypted.kind === "pong") {
|
||||
// Lease renewal is liveness, not a state-changing frame. It must
|
||||
// never queue behind a slow event commit (a long silent tool call
|
||||
// otherwise lets the runner's lease lapse). It still commits the
|
||||
// renewed expiry durably before its reply, and its admission is
|
||||
// checked against persisted authority exactly as before.
|
||||
try {
|
||||
this.#handleDecrypted(connection, decrypted);
|
||||
} catch {
|
||||
connection.close();
|
||||
}
|
||||
return;
|
||||
}
|
||||
frame = { decrypted };
|
||||
} else {
|
||||
undecryptedFrames += 1;
|
||||
frame = { raw: value };
|
||||
}
|
||||
processing = processing
|
||||
.then(() => this.#handleJson(connection, value))
|
||||
.then(() => "raw" in frame
|
||||
? this.#handleJson(connection, frame.raw).finally(() => { undecryptedFrames -= 1; })
|
||||
: this.#handleDecryptedFrame(connection, frame.decrypted))
|
||||
.catch(() => connection.close());
|
||||
const tail = processing;
|
||||
this.#connectionProcessing.set(connection, tail);
|
||||
@@ -2927,6 +2969,7 @@ export class DurablePrpControlPlane {
|
||||
void tail.then(release, release);
|
||||
},
|
||||
onClose: () => {
|
||||
this.#recordLeaseFailureIfLapsed(connection);
|
||||
this.#connections.delete(connection);
|
||||
if (connection.secureChannel !== null && this.activeRunnerConnectionCount() === 0)
|
||||
this.#runnerDisconnectedAt ??= Date.now();
|
||||
@@ -2958,6 +3001,43 @@ export class DurablePrpControlPlane {
|
||||
connection.close();
|
||||
return;
|
||||
}
|
||||
await this.#handleDecryptedFrame(connection, envelope);
|
||||
}
|
||||
|
||||
/** Decrypt one secure frame in arrival order; null when it closed the connection. */
|
||||
#decryptOnArrival(
|
||||
connection: AuthorityConnection,
|
||||
wire: unknown,
|
||||
): Record<string, unknown> | null {
|
||||
try {
|
||||
this.#store.assertWritable();
|
||||
return decryptSecureJson(connection.secureChannel!, wire);
|
||||
} catch {
|
||||
try {
|
||||
this.#store.state.malformedFrames += 1;
|
||||
this.#store.save();
|
||||
} finally {
|
||||
connection.close();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/** Handle one decrypted renewal or pong immediately, outside the frame chain. */
|
||||
#handleDecrypted(
|
||||
connection: AuthorityConnection,
|
||||
envelope: Record<string, unknown>,
|
||||
): void {
|
||||
this.#store.assertWritable();
|
||||
if (!this.#admitEnvelopeVersion(connection, envelope)) return;
|
||||
if (!this.#admitAuthenticatedFrame(connection)) return;
|
||||
if (envelope.kind === "lease_renew") this.#renewLease(connection, envelope);
|
||||
}
|
||||
|
||||
#admitEnvelopeVersion(
|
||||
connection: AuthorityConnection,
|
||||
envelope: Record<string, unknown>,
|
||||
): boolean {
|
||||
const envelopeVersion = envelope.version;
|
||||
const expectedVersion =
|
||||
connection.lease?.protocolVersion ??
|
||||
@@ -2972,8 +3052,64 @@ export class DurablePrpControlPlane {
|
||||
: envelopeVersion !== expectedVersion)
|
||||
) {
|
||||
connection.close();
|
||||
return;
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Admit every post-handshake frame against persisted authority before
|
||||
* dispatch, including lease_renew. Renewal cannot revive a revoked or
|
||||
* expired credential or bypass changes to its persisted binding. An expired
|
||||
* lease is recorded so the owner can report why the runner left.
|
||||
*/
|
||||
#admitAuthenticatedFrame(connection: AuthorityConnection): boolean {
|
||||
if (
|
||||
connection.secureChannel === null ||
|
||||
connection.lease === null ||
|
||||
canonicalJson(this.#store.state.leases[connection.lease.credentialId]) !==
|
||||
canonicalJson(connection.lease) ||
|
||||
connection.lease.revokedAt !== null ||
|
||||
connection.lease.expiresAtUnixMs <= Date.now()
|
||||
) {
|
||||
this.#recordLeaseFailureIfLapsed(connection);
|
||||
connection.close();
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
#recordLeaseFailureIfLapsed(connection: AuthorityConnection): void {
|
||||
const lease = connection.lease;
|
||||
if (lease === null || connection.secureChannel === null) return;
|
||||
const now = Date.now();
|
||||
// Only expiry is reported. A revoked credential is the expected outcome of
|
||||
// a warm handoff to a replacement runner, not a lost lease.
|
||||
if (lease.expiresAtUnixMs > now) return;
|
||||
this.#connectionLeaseFailure = {
|
||||
code: "connection_lease_expired",
|
||||
leaseId: lease.leaseId,
|
||||
expiresAtUnixMs: lease.expiresAtUnixMs,
|
||||
observedAtUnixMs: now,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Why the most recent authenticated runner connection lost its authority,
|
||||
* or null while the current connection's lease is valid. A runner whose
|
||||
* lease lapses exits without another authenticated frame, so the socket
|
||||
* close alone would otherwise look like an ordinary transport loss.
|
||||
*/
|
||||
get connectionLeaseFailure(): DurablePrpConnectionLeaseFailure | null {
|
||||
return this.#connectionLeaseFailure === null ? null : { ...this.#connectionLeaseFailure };
|
||||
}
|
||||
|
||||
async #handleDecryptedFrame(
|
||||
connection: AuthorityConnection,
|
||||
envelope: Record<string, unknown>,
|
||||
): Promise<void> {
|
||||
this.#store.assertWritable();
|
||||
if (!this.#admitEnvelopeVersion(connection, envelope)) return;
|
||||
const kind = envelope.kind;
|
||||
if (connection.secureChannel === null && kind === "auth_hello") {
|
||||
this.#authHello(connection, envelope);
|
||||
@@ -2983,20 +3119,9 @@ export class DurablePrpControlPlane {
|
||||
await this.#authResponse(connection, envelope);
|
||||
return;
|
||||
}
|
||||
// Admit every post-handshake frame against persisted authority before
|
||||
// dispatch, including lease_renew. Renewal cannot revive a revoked or
|
||||
// expired credential or bypass changes to its persisted binding.
|
||||
if (
|
||||
connection.secureChannel === null ||
|
||||
connection.lease === null ||
|
||||
canonicalJson(this.#store.state.leases[connection.lease.credentialId]) !==
|
||||
canonicalJson(connection.lease) ||
|
||||
connection.lease.revokedAt !== null ||
|
||||
connection.lease.expiresAtUnixMs <= Date.now()
|
||||
) {
|
||||
connection.close();
|
||||
return;
|
||||
}
|
||||
if (!this.#admitAuthenticatedFrame(connection)) return;
|
||||
// After admission the lease is non-null; keep the narrowed reference.
|
||||
if (connection.lease === null) return;
|
||||
if (kind === "lease_renew") {
|
||||
this.#renewLease(connection, envelope);
|
||||
return;
|
||||
@@ -3108,15 +3233,21 @@ export class DurablePrpControlPlane {
|
||||
// Repeating a request after a lost reply replays the persisted expiry.
|
||||
// It never extends a credential twice for the same observed generation.
|
||||
if (expectedExpiry === lease.expiresAtUnixMs) {
|
||||
const candidate = structuredClone(this.#store.state);
|
||||
const renewed = candidate.leases[lease.credentialId]!;
|
||||
renewed.expiresAtUnixMs = Math.max(
|
||||
const expiresAtUnixMs = Math.max(
|
||||
lease.expiresAtUnixMs,
|
||||
Date.now() + this.#connectionLeaseTtlMs,
|
||||
);
|
||||
renewed.expiresAt = new Date(renewed.expiresAtUnixMs).toISOString();
|
||||
candidate.lastLeaseExpiresAt = renewed.expiresAt;
|
||||
this.#store.commit(candidate);
|
||||
const renewed = {
|
||||
...structuredClone(lease),
|
||||
expiresAtUnixMs,
|
||||
expiresAt: new Date(expiresAtUnixMs).toISOString(),
|
||||
};
|
||||
// Renewal runs outside the frame chain, so an event's staged write can
|
||||
// be in flight. save() absorbs that admitted event into this write; an
|
||||
// explicit commit() would invalidate it and close a healthy connection.
|
||||
this.#store.state.leases[lease.credentialId] = renewed;
|
||||
this.#store.state.lastLeaseExpiresAt = renewed.expiresAt;
|
||||
this.#store.save();
|
||||
connection.lease = this.#store.state.leases[lease.credentialId]!;
|
||||
}
|
||||
this.#sendJson(connection,
|
||||
@@ -3642,6 +3773,7 @@ export class DurablePrpControlPlane {
|
||||
}
|
||||
|
||||
this.#store.state.connectionCount += 1;
|
||||
this.#connectionLeaseFailure = null;
|
||||
this.#store.state.lastLeaseId = lease.leaseId;
|
||||
this.#store.state.lastLeaseExpiresAt = lease.expiresAt;
|
||||
|
||||
|
||||
@@ -45,6 +45,7 @@ export {
|
||||
type StagedRunAttachment,
|
||||
type DurablePrpControlPlaneOptions,
|
||||
type DurablePrpCheckpoint,
|
||||
type DurablePrpConnectionLeaseFailure,
|
||||
decodeDurablePrpCheckpoint,
|
||||
captureLegacyDurablePrpCheckpoint,
|
||||
inspectDurablePrpCheckpointHeader,
|
||||
|
||||
@@ -455,6 +455,69 @@ describe("runner provider-ingress WebSocket wire", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["refused", "unauthenticated"] as const)("names an expired runner connection lease in the run log and terminal error when reconnect is %s", async reconnect => {
|
||||
vi.useFakeTimers();
|
||||
const first = new FakeSocket();
|
||||
const create = sockets.create;
|
||||
let opens = 0;
|
||||
sockets.create = () => {
|
||||
opens += 1;
|
||||
if (opens === 1) return openingSocket(first);
|
||||
// An open socket whose PRP authentication never completes, as when the
|
||||
// controller refuses the runner's lapsed credential.
|
||||
if (reconnect === "unauthenticated") return openingSocket();
|
||||
const refused = new FakeSocket();
|
||||
refused.readyState = WebSocket.CONNECTING;
|
||||
queueMicrotask(() => {
|
||||
refused.readyState = WebSocket.CLOSED;
|
||||
refused.emit("error", new Error("fixture runner exited"));
|
||||
});
|
||||
return refused;
|
||||
};
|
||||
const endpoint = {
|
||||
kind: "authenticated_websocket" as const,
|
||||
websocketUrl: "wss://preview.invalid/api/runner/v1/connect/run-lease-expired",
|
||||
secretHeaders: [],
|
||||
generation: "generation-lease-expired",
|
||||
refresh: vi.fn(async () => endpoint),
|
||||
close: vi.fn(async () => undefined),
|
||||
};
|
||||
let leaseFailure: { code: string } | null = null;
|
||||
const states: Array<[string, string | undefined]> = [];
|
||||
const handle = connectRunnerPrpIngress({
|
||||
authority: {
|
||||
attachWireConnection: vi.fn(() => {
|
||||
const authenticated = opens === 1;
|
||||
return { isAuthenticated: () => authenticated };
|
||||
}),
|
||||
activeRunnerConnectionCount: () => 0,
|
||||
get connectionLeaseFailure() { return leaseFailure; },
|
||||
} as unknown as DurablePrpControlPlane,
|
||||
endpoint,
|
||||
recoveryGraceMs: 10_000,
|
||||
onStateChange: (state, code) => states.push([state, code]),
|
||||
});
|
||||
const failure = vi.fn();
|
||||
void handle.failure.catch(failure);
|
||||
try {
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
await handle.ready;
|
||||
// The runner's lease lapsed; it exits and the socket closes.
|
||||
leaseFailure = { code: "connection_lease_expired" };
|
||||
first.close(1006);
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(states).toContainEqual(["reconnecting", "runner_connection_lease_expired"]);
|
||||
await vi.advanceTimersByTimeAsync(20_000);
|
||||
expect(failure).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ code: "runner_connection_lease_expired" }),
|
||||
);
|
||||
} finally {
|
||||
await handle.close();
|
||||
sockets.create = create;
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps a quiet but live runner attached through a long silent tool call", async () => {
|
||||
vi.useFakeTimers();
|
||||
const live = new FakeSocket();
|
||||
|
||||
@@ -21,7 +21,8 @@ const DEFAULT_LIVENESS_TIMEOUT_MS = 90_000;
|
||||
|
||||
export type RunnerIngressFailureCode =
|
||||
| "runner_ingress_unavailable"
|
||||
| "runner_ingress_auth_failed";
|
||||
| "runner_ingress_auth_failed"
|
||||
| "runner_connection_lease_expired";
|
||||
|
||||
export class RunnerIngressConnectionError extends Error {
|
||||
constructor(
|
||||
@@ -354,6 +355,10 @@ export function connectRunnerPrpIngress(input: {
|
||||
const loop = (async () => {
|
||||
let attempt = 0;
|
||||
let recoveryDeadline = startupDeadline;
|
||||
// A runner whose connection lease lapsed exits without another
|
||||
// authenticated frame. Keep the authority's reason so the run log and the
|
||||
// terminal error name it instead of a generic transport loss.
|
||||
let leaseFailureCode: RunnerIngressFailureCode | null = null;
|
||||
while (!abort.signal.aborted) {
|
||||
try {
|
||||
input.onStateChange?.(attempt === 0 ? "connecting" : "reconnecting");
|
||||
@@ -387,6 +392,7 @@ export function connectRunnerPrpIngress(input: {
|
||||
}
|
||||
input.onStateChange?.("authenticated");
|
||||
attempt = 0;
|
||||
leaseFailureCode = null;
|
||||
const stopLiveness = opened.wire.monitorLiveness({
|
||||
intervalMs: input.livenessIntervalMs ?? DEFAULT_LIVENESS_INTERVAL_MS,
|
||||
timeoutMs: input.livenessTimeoutMs ?? DEFAULT_LIVENESS_TIMEOUT_MS,
|
||||
@@ -397,6 +403,11 @@ export function connectRunnerPrpIngress(input: {
|
||||
stopLiveness();
|
||||
}
|
||||
recoveryDeadline = Date.now() + recoveryGraceMs;
|
||||
const leaseFailure = input.authority.connectionLeaseFailure ?? null;
|
||||
if (leaseFailure) {
|
||||
leaseFailureCode = "runner_connection_lease_expired";
|
||||
input.onStateChange?.("reconnecting", leaseFailureCode);
|
||||
}
|
||||
} catch (error) {
|
||||
// close() owns intentional shutdown. openEndpoint rejects its pending
|
||||
// handshake when the abort signal fires, but that cancellation is not
|
||||
@@ -431,8 +442,16 @@ export function connectRunnerPrpIngress(input: {
|
||||
}
|
||||
}
|
||||
if (Date.now() >= recoveryDeadline) {
|
||||
// A lapsed lease is the root cause even when a later reconnect
|
||||
// fails authentication; keep that failure as the cause.
|
||||
const terminal =
|
||||
failure instanceof RunnerIngressConnectionError
|
||||
leaseFailureCode !== null
|
||||
? new RunnerIngressConnectionError(
|
||||
leaseFailureCode,
|
||||
"The runner's connection lease expired before it was renewed.",
|
||||
{ cause: failure },
|
||||
)
|
||||
: failure instanceof RunnerIngressConnectionError
|
||||
? failure
|
||||
: new RunnerIngressConnectionError(
|
||||
statusCode === 401 || statusCode === 403
|
||||
|
||||
Reference in new issue
Block a user