diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index d4e3a4effea..282d74a83bc 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -188,12 +188,16 @@ export class NativeSessionProtocolIntegrityError extends Error { constructor( readonly reason: | "semantic_input_digest_mismatch" - | "source_event_replay_conflict", + | "source_event_replay_conflict" + | "runner_event_commit_stalled", + readonly detail?: string, ) { super( reason === "semantic_input_digest_mismatch" ? "native_event_replay_conflict: authenticated runner semantic input failed integrity validation; automatic recovery is stopped." - : "native_event_replay_conflict: authenticated runner event conflicts with committed history; automatic recovery is stopped.", + : reason === "runner_event_commit_stalled" + ? `native_event_replay_conflict: runner_event_commit_stalled: runner events kept failing to commit without progress (last cause: ${detail ?? "unknown"}); automatic recovery is stopped.` + : "native_event_replay_conflict: authenticated runner event conflicts with committed history; automatic recovery is stopped.", ); this.name = "NativeSessionProtocolIntegrityError"; } diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts index 1332df7ddf6..699d5bdda26 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts @@ -1345,6 +1345,200 @@ it.each(["success", "integrity_failure"] as const)("ignores a replaced connectio } }); +it("fails closed with the schema issue when the runner keeps redelivering an invalid event (restart chaos t01)", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-event-schema-stall-")); + const commits = vi.fn(async () => undefined); + const onProtocolIntegrityError = vi.fn(); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: commits, onProtocolIntegrityError }); + const envelope = semanticInputEvent(1); + const event = envelope.payload as Record; + event.sourceEventId = "invalid-write-target-1"; + event.eventType = "auth.credentials.updated"; + event.payload = { schema: "paperclip.auth.credentials.updated.v1", provider: "codex", + bindingId: "harness-auth", revision: 1, detection: "credential_store" }; + delete event.schemaVersion; + event.priority = 7; + const now = vi.spyOn(Date, "now"); + let clock = 1_000_000; + now.mockImplementation(() => clock); + try { + await core.start(); + let lease: string | null = null; + for (let attempt = 1; attempt <= 20; attempt += 1) { + const client: AuthenticatedClient = (await authenticate(core, lease ?? core.issueBootstrapTicket()))!; + lease = client.leaseToken ?? lease; + sendSecure(client, envelope); + await vi.waitFor(() => expect(client.socket.destroyed || client.socket.readyState === "closed" + || onProtocolIntegrityError.mock.calls.length > 0).toBe(true)); + clock += 3_000; + client.socket.destroy(); + } + expect(commits).not.toHaveBeenCalled(); + expect(onProtocolIntegrityError).toHaveBeenCalledOnce(); + const failure = onProtocolIntegrityError.mock.calls[0]![0] as { reason: string; detail: string }; + expect(failure.reason).toBe("runner_event_commit_stalled"); + expect(failure.detail).toMatch(/^event_schema_invalid:\/\S* at source seq 1$/); + } finally { + now.mockRestore(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + +it("fails closed at once when the journal cannot record a runner event", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-event-journal-failure-")); + const onProtocolIntegrityError = vi.fn(); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: vi.fn(async () => undefined), onProtocolIntegrityError }); + const envelope = semanticInputEvent(1); + const event = envelope.payload as Record; + event.sourceEventId = "journal-failure-1"; + event.eventType = "auth.credentials.updated"; + event.payload = { schema: "paperclip.auth.credentials.updated.v1", provider: "codex", + bindingId: "harness-auth", revision: 1, detection: "credential_store" }; + try { + await core.start(); + const client: AuthenticatedClient = (await authenticate(core, core.issueBootstrapTicket()))!; + const journal = core.store as unknown as { commit: (...args: unknown[]) => void; commitEvent: (...args: unknown[]) => Promise }; + const diskFull = Object.assign(new Error("no space on /private/disk"), { code: "ENOSPC" }); + vi.spyOn(journal, "commit").mockImplementation(() => { throw diskFull; }); + vi.spyOn(journal, "commitEvent").mockRejectedValue(diskFull); + sendSecure(client, envelope); + await vi.waitFor(() => expect(onProtocolIntegrityError).toHaveBeenCalledOnce()); + const failure = onProtocolIntegrityError.mock.calls[0]![0] as Error & { reason: string; detail: string }; + expect(failure).toMatchObject({ reason: "runner_event_commit_stalled", detail: "journal_write_failed:ENOSPC at source seq 1" }); + expect(failure.message).not.toContain("/private/disk"); + client.socket.destroy(); + } finally { + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + +it("fails closed when checkpoint storage keeps failing before a reconnect authenticates", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-checkpoint-stall-")); + const onProtocolIntegrityError = vi.fn(); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: vi.fn(async () => undefined), onProtocolIntegrityError }); + let checkpointUp = true; + core.setCheckpointPersistence(async () => { + if (!checkpointUp) throw Object.assign(new Error("connect to private-db-host refused"), { code: "ECONNREFUSED" }); + }); + const now = vi.spyOn(Date, "now"); + let clock = 1_000_000; + now.mockImplementation(() => clock); + try { + await core.start(); + const tickets = Array.from({ length: 20 }, () => core.issueBootstrapTicket(60_000)); + await core.flushCheckpoint(); + checkpointUp = false; + for (const ticket of tickets) { + const client = await authenticate(core, ticket); + client?.socket.destroy(); + clock += 3_000; + if (onProtocolIntegrityError.mock.calls.length > 0) break; + } + expect(onProtocolIntegrityError).toHaveBeenCalledOnce(); + const failure = onProtocolIntegrityError.mock.calls[0]![0] as Error & { reason: string; detail: string }; + expect(failure.reason).toBe("runner_event_commit_stalled"); + expect(failure.detail).toMatch(/^checkpoint_flush_failed:ECONNREFUSED at source seq 1$/); + expect(failure.message).not.toContain("private-db-host"); + } finally { + now.mockRestore(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + +it.each([ + { label: "an out-of-order event", seq: 5, failCommit: false, stepMs: 3_000, failsClosed: true, + detail: "event_sequence_mismatch at source seq 5" }, + { label: "20 failures inside 45 s", seq: 1, failCommit: true, stepMs: 1_000, failsClosed: false, detail: null }, +])("bounds commit-stall reconnects for $label", async ({ seq, failCommit, stepMs, failsClosed, detail }) => { + const root = mkdtempSync(resolve(tmpdir(), "runner-event-stall-bound-")); + const commits = vi.fn(async () => { + if (failCommit) throw Object.assign(new Error("transient"), { code: "ECONNRESET" }); + }); + const onProtocolIntegrityError = vi.fn(); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: commits, onProtocolIntegrityError }); + const envelope = semanticInputEvent(seq); + const event = envelope.payload as Record; + event.sourceEventId = `stall-bound-${seq}`; + event.eventType = "auth.credentials.updated"; + event.payload = { schema: "paperclip.auth.credentials.updated.v1", provider: "codex", + bindingId: "harness-auth", revision: 1, detection: "credential_store" }; + const now = vi.spyOn(Date, "now"); + let clock = 1_000_000; + now.mockImplementation(() => clock); + try { + await core.start(); + let lease: string | null = null; + for (let attempt = 1; attempt <= 20; attempt += 1) { + const client: AuthenticatedClient = (await authenticate(core, lease ?? core.issueBootstrapTicket()))!; + lease = client.leaseToken ?? lease; + sendSecure(client, envelope); + if (failCommit) await vi.waitFor(() => expect(commits).toHaveBeenCalledTimes(attempt)); + await new Promise(resolveTurn => setTimeout(resolveTurn, 20)); + clock += stepMs; + client.socket.destroy(); + } + if (failsClosed) { + expect(onProtocolIntegrityError).toHaveBeenCalledOnce(); + expect(onProtocolIntegrityError.mock.calls[0]![0]).toMatchObject({ reason: "runner_event_commit_stalled", detail }); + } else { + expect(onProtocolIntegrityError).not.toHaveBeenCalled(); + } + } finally { + now.mockRestore(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + +it("fails closed when runner events keep failing to commit without progress", async () => { + const root = mkdtempSync(resolve(tmpdir(), "runner-event-delivery-stall-")); + const commits = vi.fn(async () => { + throw Object.assign(new Error("connection reset by private-host-fixture"), { code: "ECONNRESET" }); + }); + const onProtocolIntegrityError = vi.fn(); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, + expectedRunnerVersion, expectedRunnerDigest, onCommittedEvent: commits, onProtocolIntegrityError }); + const envelope = semanticInputEvent(1); + const event = envelope.payload as Record; + event.sourceEventId = "stalled-delivery-1"; + event.eventType = "auth.credentials.updated"; + event.payload = { schema: "paperclip.auth.credentials.updated.v1", provider: "codex", + bindingId: "harness-auth", revision: 1, detection: "credential_store" }; + const now = vi.spyOn(Date, "now"); + let clock = 1_000_000; + now.mockImplementation(() => clock); + try { + await core.start(); + let lease: string | null = null; + for (let attempt = 1; attempt <= 20; attempt += 1) { + const client: AuthenticatedClient = (await authenticate(core, lease ?? core.issueBootstrapTicket()))!; + lease = client.leaseToken ?? lease; + sendSecure(client, envelope); + await vi.waitFor(() => expect(commits).toHaveBeenCalledTimes(attempt)); + await new Promise(resolveTurn => setImmediate(resolveTurn)); + if (attempt < 20) expect(onProtocolIntegrityError).not.toHaveBeenCalled(); + clock += 3_000; + client.socket.destroy(); + } + expect(onProtocolIntegrityError).toHaveBeenCalledOnce(); + const failure = onProtocolIntegrityError.mock.calls[0]![0] as Error; + expect(failure).toMatchObject({ reason: "runner_event_commit_stalled", detail: "ECONNRESET at source seq 1" }); + expect(failure.message).toContain("runner_event_commit_stalled"); + expect(failure.message).not.toContain("private-host-fixture"); + } finally { + now.mockRestore(); + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + function renewalRequest(client: AuthenticatedClient, expiresAt: number): Record { return { protocol: "paperclip.runner", version: client.welcome.version, diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index e2e61d45b72..cf85f3ffbe6 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -73,6 +73,8 @@ class CommandJournalLimitError extends Error { // polling turn regains the event loop. Match the transport's explicit deferred // event bound so a valid burst is not compacted before it can be observed. const maxCommittedEventWindow = 4_096; +const EVENT_DELIVERY_STALL_FAILURES = 20; +const EVENT_DELIVERY_STALL_MS = 45_000; const maxStateBytes = 256 * 1024 * 1024; const authChallengeTtlMs = 5_000; const stableIdPattern = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,239}$/; @@ -870,6 +872,19 @@ function validStoredWarmTransition(state: StoredCoreState): boolean { } } +function eventSchemaIssuePaths(issues: readonly { path?: unknown }[] | undefined): string { + const paths = [...new Set((issues ?? []).map(issue => typeof issue.path === "string" ? issue.path : "") + .map(path => path.replace(/[^A-Za-z0-9_./-]/g, "_").slice(0, 80)).filter(Boolean))]; + return paths.length ? paths.slice(0, 3).join(",") : "unknown"; +} + +function eventCommitFailureCause(error: unknown): string { + const code = isRecord(error) ? error.code : undefined; + if (typeof code === "string" && /^[A-Za-z0-9_.:-]{1,80}$/.test(code)) return code; + const name = error instanceof Error ? error.name : typeof error; + return /^[A-Za-z0-9_.:-]{1,80}$/.test(name) ? name : "unknown_error"; +} + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } @@ -2040,6 +2055,7 @@ export class DurablePrpControlPlane { #onProtocolIntegrityError?: DurablePrpControlPlaneOptions["onProtocolIntegrityError"]; #onAuthenticatedConnection?: DurablePrpControlPlaneOptions["onAuthenticatedConnection"]; #protocolIntegrityError: NativeSessionProtocolIntegrityError | null = null; + #eventDeliveryStall: { runId: string; ackedSourceSeq: number; since: number; failures: number } | null = null; #connectionLeaseTtlMs: number; #persistCheckpoint: ((checkpoint: DurablePrpCheckpoint) => Promise) | null = null; #persistedRevision = -1; @@ -2211,10 +2227,15 @@ export class DurablePrpControlPlane { } const checkpoint = this.flushCheckpoint(); const frame = structuredClone(value); + let checkpointFailure: unknown = null; const sending = (this.#sendQueues.get(connection) ?? Promise.resolve()) - .then(() => checkpoint) + .then(() => checkpoint.catch(error => { checkpointFailure = error ?? new Error("checkpoint_failed"); throw error; })) .then(() => connection.sendJson(frame)) - .catch(() => connection.close()); + .catch(() => { + if (checkpointFailure !== null && this.#protocolIntegrityError === null) { + this.#eventCommitFailed(connection, `checkpoint_flush_failed:${eventCommitFailureCause(checkpointFailure)}`); + } else connection.close(); + }); this.#sendQueues.set(connection, sending); void sending.finally(() => { if (this.#sendQueues.get(connection) === sending) this.#sendQueues.delete(connection); @@ -3806,7 +3827,12 @@ export class DurablePrpControlPlane { this.#store.state.commandDeliveryCounts[command.commandId] = (this.#store.state.commandDeliveryCounts[command.commandId] ?? 0) + 1; } - this.#store.save(); + try { + this.#store.save(); + } catch (error) { + this.#failEventCommit(connection, `journal_write_failed:${eventCommitFailureCause(error)}`); + return; + } this.#sendJson(connection, { protocol, version: lease.protocolVersion, @@ -4124,6 +4150,27 @@ export class DurablePrpControlPlane { ); } + #eventCommitFailed(connection: AuthorityConnection, cause: string, sourceSeq?: number, now = Date.now()): void { + const acked = this.#store.state.ackedSourceSeq; + const runId = this.#identity.runId; + const stall = this.#eventDeliveryStall; + if (stall === null || stall.ackedSourceSeq !== acked || stall.runId !== runId) { + this.#eventDeliveryStall = { runId, ackedSourceSeq: acked, since: now, failures: 1 }; + } else { + stall.failures += 1; + if (stall.failures >= EVENT_DELIVERY_STALL_FAILURES && now - stall.since >= EVENT_DELIVERY_STALL_MS) { + this.#failEventCommit(connection, cause, sourceSeq); + return; + } + } + connection.close(); + } + + #failEventCommit(connection: AuthorityConnection, cause: string, sourceSeq?: number): void { + this.#failProtocolIntegrity(connection, new NativeSessionProtocolIntegrityError("runner_event_commit_stalled", + `${cause} at source seq ${sourceSeq ?? this.#store.state.ackedSourceSeq + 1}`)); + } + #failProtocolIntegrity( connection: AuthorityConnection, error: NativeSessionProtocolIntegrityError, @@ -4164,7 +4211,7 @@ export class DurablePrpControlPlane { } const validated = validatePrpEvent(envelope.payload); if (!validated.ok) { - connection.close(); + this.#eventCommitFailed(connection, `event_schema_invalid:${eventSchemaIssuePaths(validated.issues)}`); return; } const event = validated.event; @@ -4218,7 +4265,7 @@ export class DurablePrpControlPlane { ? sourceSeq !== this.#store.state.ackedSourceSeq + 1 : sourceSeq !== existing.sourceSeq ) { - connection.close(); + this.#eventCommitFailed(connection, "event_sequence_mismatch", sourceSeq); return; } if ( @@ -4277,7 +4324,7 @@ export class DurablePrpControlPlane { if (error instanceof NativeSessionProtocolIntegrityError) { this.#failProtocolIntegrity(connection, error); } else { - connection.close(); + this.#eventCommitFailed(connection, eventCommitFailureCause(error), sourceSeq); } return; } @@ -4324,13 +4371,28 @@ export class DurablePrpControlPlane { // turn before readers react or a replacement owner adopts their cursor. const candidate = buildCandidate(this.#store.state); if (candidate === null) { - connection.close(); + this.#eventCommitFailed(connection, "journal_sequence_not_contiguous", sourceSeq); + return; + } + try { + this.#store.commit(candidate, !isStreamingEvent(event)); + } catch (error) { + this.#failEventCommit(connection, `journal_write_failed:${eventCommitFailureCause(error)}`, sourceSeq); + return; + } + } else { + let committed: boolean; + try { + committed = await this.#store.commitEvent(buildCandidate, isCurrentOwner, !isStreamingEvent(event)); + } catch (error) { + if (isCurrentOwner()) this.#failEventCommit(connection, `journal_write_failed:${eventCommitFailureCause(error)}`, sourceSeq); + else connection.close(); + return; + } + if (!committed) { + if (isCurrentOwner()) this.#eventCommitFailed(connection, "journal_commit_refused", sourceSeq); return; } - this.#store.commit(candidate, !isStreamingEvent(event)); - } else if (!await this.#store.commitEvent(buildCandidate, isCurrentOwner, !isStreamingEvent(event))) { - if (isCurrentOwner()) connection.close(); - return; } if (!isCurrentOwner()) return; @@ -4350,7 +4412,14 @@ export class DurablePrpControlPlane { if (ownsLiveDispatch) this.#pendingSemanticCalls.add(liveCommandId!); let dispatched = false; try { - await this.flushCheckpoint(); + try { + await this.flushCheckpoint(); + } catch (error) { + if (this.#eventOwner === connection && this.#identity === identity) { + this.#eventCommitFailed(connection, `checkpoint_flush_failed:${eventCommitFailureCause(error)}`, sourceSeq); + } + return; + } if (this.#eventOwner !== connection || this.#identity !== identity) return; if (!this.#retainsSourceInputs() && ownsLiveDispatch && liveCall && liveCommandId && this.#onSemanticToolInput && !this.#getCommand(liveCommandId)) { const call = liveCall;