mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-05 21:02:34 +01:00
fix(chat): retain receiver ownership during runtime construction (#18930)
## Thinking Path > - Paperclip manages agents and their work. > - Chat receivers hold a durable lease before they accept messages. > - Runtime construction can outlast a lease renewal interval. > - Renewal previously required an attached runtime before construction had returned. > - That check could revoke the same receiver that was still starting. > - This change uses the exact pending initialization as proof during construction. > - The existing durable token and endpoint checks still fence every renewal. ## Linked Issues or Issue Description Related: #18899 exposed the existing receiver failure in its chat CI shard. This is a separate lifecycle repair. #18795 provides the relevant production and sleep fixture context. Searches found no overlapping open initialization repair. **What happened?** The receiver acquired its lease and started its renewal timer. The service then awaited credentials, command preparation and runtime construction before attaching the runtime to its context. A renewal during that interval treated the missing attachment as lost ownership. Configuration later failed with a Gateway ownership error, although its original durable token was still current. **Expected behavior** The exact pending initialization can retain its own lease while construction runs. Missing attachment alone must grant no ownership. A changed token, paused or removed endpoint, changed generation or credentials, and shutdown must still fence the receiver. **Steps to reproduce** 1. Use the real configure and PostgreSQL ownership producers with a 60-second lease and 50-millisecond renewal. 2. Hold the actual runtime construction call for 200 milliseconds. 3. Release construction. Untouched code returns the Gateway ownership error. 4. With the same construction delay and a five-second renewal control, configuration succeeds. **Paperclip version or commit** The causal regression fails on unchanged production source at `e7549ab613`. The original CI error was observed in the chat shard of #18899. That CI run did not record the precise setup and renewal order, so this local proof does not establish its historical scheduling. ## What Changed - Store the runtime context identity in the existing pending initialization entry. - Allow pre-attachment renewal only for that exact tracked initialization and current local ownership. - Recheck local ownership inside the existing renewal transaction. Preserve company, endpoint, generation, credential and original-token checks. - Before SDK listener initialization, wait for any older renewal, recheck local ownership, then obtain fresh durable proof. Preserve the existing checks after initialization. - Keep attached runtime identity checks, lease deadlines, renewal intervals, shutdown and listener retirement unchanged. - Add actual configure and database race controls. Document the shared Discord and Photon receiver contract. ## Verification - Untouched production red: the actual construction gate crosses a renewal and returns the same Gateway ownership error. The otherwise identical slower-renewal control passes. No lease, exception or provider outcome is injected in these positive cases. - A subsequent review found a second real race: final runtime removal did not prove that SDK initialization had never started. Four actual initialization-hook controls failed before the pre-initialize fence. The final matrix asserts zero hook calls for lost or changed authority, including a foreign-token control with no periodic renewal due. An attempted cross-company endpoint move is rejected by the existing database foreign key; the valid original initialization still succeeds. - A third actual-producer red holds completion of a real committed renewal, changes the durable token, and proves that joining that old promise could incorrectly accept configuration. The corrected startup check waits for that renewal and then obtains fresh proof; it cannot credit the earlier result alone. - Final relevant full Discord registration group, takeover/callback cases, buffered host-pause recovery, failed construction, credential-fenced pause and shared Photon consumers: 34 passed across two files; 1,840 other cases were deselected. Four executed cases cover Photon transport recovery, failed authentication, duplicate allocation ownership and checkpoint fencing. The tests use the real configure/database producers with fake provider transports and observe SDK initialization; they do not prove physical network behavior. - All server type layers, the module-boundary gate, and production server/vendored Runner build pass at exact final `38cf82ec114bd12c0aa7bbc125ed72b1ae941b4e`. The production build stamp records this exact commit. - Earlier fixture mistakes are preserved separately: a test initially changed every lease key, including the credential lease, and another attempted an invalid cross-company row move. Neither is counted as a production behavior red. The corrected controls target the Gateway key and assert the existing company constraint. - An initial isolated dependency graph resolved an older shared telemetry build. Correct workspace links restored the all-layer typecheck without changing telemetry source. - Full repository checks were not run locally for this narrow slice. Fresh current-head CI and substantive reviews remain required. No hosted task, provider work, fleet action or deployment was performed. ## Risks The ownership helper is shared by Discord and Photon. The selected tests cover both consumers. The pending entry grants only permission to renew the original token; it grants no message, callback or publication authority and adopts no listener. Missing or changed initialization identity fails closed. There is no schema change, historical backfill, new worker or additional idle scan. ## Model Used OpenAI `gpt-6.1-sol` in Codex, with reasoning, repository inspection, source editing, shell execution and test tools. The host did not report the context-window size. A separate coordinator reviews the source before delivery. ## 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 --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
c6fe9b143c
commit
0dedbf23c3
4 files changed
+132
-9
No files matched your search
@@ -8800,6 +8800,96 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["owned", "slow-renewal", "lost-token", "foreign-token", "foreign-token-before-renewal", "paused", "removed", "replacement", "credential-change", "foreign-company", "shutdown"] as const)("fences receiver construction across renewal: %s", async fault => {
|
||||
const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 60_000, discordGatewayLeaseRenewalIntervalMs: fault === "slow-renewal" || fault === "foreign-token-before-renewal" ? 5000 : 50 });
|
||||
const listenerStarted = vi.fn(async () => {});
|
||||
f.runtime.initializeHook = listenerStarted;
|
||||
let release!: () => void;
|
||||
let entered!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const constructing = new Promise<void>(resolve => { entered = resolve; });
|
||||
const replace = f.runtime.replaceEndpoint.bind(f.runtime);
|
||||
const replacement = vi.spyOn(f.runtime, "replaceEndpoint").mockImplementation(async options => {
|
||||
entered(); await gate; return replace(options);
|
||||
});
|
||||
const configured = f.configure().then(value => ({ value }), error => ({ error }));
|
||||
const foreignToken = randomUUID();
|
||||
let shutdown: Promise<void> | undefined;
|
||||
try {
|
||||
await constructing;
|
||||
if (fault === "lost-token") await db.delete(chatEndpointLeases).where(and(eq(chatEndpointLeases.endpointId, f.endpoint.id), eq(chatEndpointLeases.leaseKey, "discord_gateway_runtime")));
|
||||
if (fault === "foreign-token" || fault === "foreign-token-before-renewal") await db.update(chatEndpointLeases).set({ token: foreignToken }).where(and(eq(chatEndpointLeases.endpointId, f.endpoint.id), eq(chatEndpointLeases.leaseKey, "discord_gateway_runtime")));
|
||||
if (fault === "paused") await db.update(chatEndpoints).set({ status: "paused" }).where(eq(chatEndpoints.id, f.endpoint.id));
|
||||
if (fault === "removed") await db.delete(chatEndpoints).where(eq(chatEndpoints.id, f.endpoint.id));
|
||||
if (fault === "replacement") await db.update(chatEndpoints).set({ setup: sql`${chatEndpoints.setup} || jsonb_build_object('runtimeGeneration', (${chatEndpoints.setup}->>'runtimeGeneration')::integer + 1)` }).where(eq(chatEndpoints.id, f.endpoint.id));
|
||||
if (fault === "foreign-company") {
|
||||
const foreign = await seedCompany();
|
||||
await expect(db.update(chatEndpoints).set({ companyId: foreign.companyId }).where(eq(chatEndpoints.id, f.endpoint.id)))
|
||||
.rejects.toMatchObject({ cause: { code: "23503" } });
|
||||
}
|
||||
if (fault === "credential-change") await db.update(toolConnections).set({ credentialSecretRefs: [] }).where(eq(toolConnections.id, f.endpoint.connectionId));
|
||||
if (fault === "shutdown") shutdown = f.service.shutdown();
|
||||
await new Promise(resolve => setTimeout(resolve, 200));
|
||||
release();
|
||||
const result = await configured;
|
||||
const [lease] = await db.select().from(chatEndpointLeases).where(and(eq(chatEndpointLeases.endpointId, f.endpoint.id), eq(chatEndpointLeases.leaseKey, "discord_gateway_runtime")));
|
||||
if (fault === "owned" || fault === "slow-renewal" || fault === "foreign-company") {
|
||||
expect(result).not.toHaveProperty("error");
|
||||
expect(listenerStarted).toHaveBeenCalledOnce();
|
||||
expect(f.runtime.get(f.endpoint.id)).not.toBeNull();
|
||||
expect(lease).toBeDefined();
|
||||
} else {
|
||||
expect(result).toHaveProperty("error");
|
||||
expect(listenerStarted).not.toHaveBeenCalled();
|
||||
expect(f.runtime.get(f.endpoint.id)).toBeNull();
|
||||
if (fault === "foreign-token" || fault === "foreign-token-before-renewal") expect(lease?.token).toBe(foreignToken);
|
||||
else expect(lease).toBeUndefined();
|
||||
}
|
||||
} finally {
|
||||
release(); await configured; await shutdown; replacement.mockRestore();
|
||||
if (fault === "removed") await f.service.shutdown();
|
||||
else await retireRegistrationFixture(f.service, f.endpoint.id);
|
||||
}
|
||||
});
|
||||
|
||||
it("rechecks durable ownership after an earlier renewal finishes before listener start", async () => {
|
||||
const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 60_000, discordGatewayLeaseRenewalIntervalMs: 50 });
|
||||
const listenerStarted = vi.fn(async () => {}); f.runtime.initializeHook = listenerStarted;
|
||||
let releaseConstruction!: () => void, constructionEntered!: () => void, releaseRenewal!: () => void, renewalEntered!: () => void;
|
||||
const construction = new Promise<void>(resolve => { releaseConstruction = resolve; });
|
||||
const enteredConstruction = new Promise<void>(resolve => { constructionEntered = resolve; });
|
||||
const renewal = new Promise<void>(resolve => { releaseRenewal = resolve; });
|
||||
const enteredRenewal = new Promise<void>(resolve => { renewalEntered = resolve; });
|
||||
let constructionPending = false;
|
||||
const replace = f.runtime.replaceEndpoint.bind(f.runtime);
|
||||
const replacement = vi.spyOn(f.runtime, "replaceEndpoint").mockImplementation(async options => {
|
||||
constructionPending = true; constructionEntered(); await construction; return replace(options);
|
||||
});
|
||||
let gated = false;
|
||||
const transact = db.transaction.bind(db);
|
||||
const transaction = vi.spyOn(db, "transaction").mockImplementation((async (...args: Parameters<typeof db.transaction>) => {
|
||||
const result = await transact(...args);
|
||||
if (result === true && constructionPending && !gated) { gated = true; renewalEntered(); await renewal; }
|
||||
return result;
|
||||
}) as typeof db.transaction);
|
||||
const configured = f.configure().then(value => ({ value }), error => ({ error }));
|
||||
const foreignToken = randomUUID();
|
||||
try {
|
||||
await enteredConstruction; await enteredRenewal;
|
||||
await db.update(chatEndpointLeases).set({ token: foreignToken }).where(and(eq(chatEndpointLeases.endpointId, f.endpoint.id), eq(chatEndpointLeases.leaseKey, "discord_gateway_runtime")));
|
||||
releaseConstruction();
|
||||
await vi.waitFor(() => expect(f.runtime.get(f.endpoint.id)).not.toBeNull());
|
||||
releaseRenewal();
|
||||
expect(await configured).toHaveProperty("error");
|
||||
expect(listenerStarted).not.toHaveBeenCalled();
|
||||
const [lease] = await db.select().from(chatEndpointLeases).where(and(eq(chatEndpointLeases.endpointId, f.endpoint.id), eq(chatEndpointLeases.leaseKey, "discord_gateway_runtime")));
|
||||
expect(lease?.token).toBe(foreignToken);
|
||||
} finally {
|
||||
releaseConstruction(); releaseRenewal(); await configured; transaction.mockRestore(); replacement.mockRestore();
|
||||
await retireRegistrationFixture(f.service, f.endpoint.id);
|
||||
}
|
||||
});
|
||||
|
||||
it("restarts a receiver after token loss only after the old listener has stopped", async () => {
|
||||
const f = await registrationFixture("success", { discordGatewayLeaseTtlMs: 60_000, discordGatewayLeaseRenewalIntervalMs: 50 });
|
||||
const onError = vi.fn();
|
||||
|
||||
@@ -513,6 +513,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
const runtimeInitializations = new Map<
|
||||
string,
|
||||
{
|
||||
context: RuntimeContext;
|
||||
discordGatewayOwned: boolean;
|
||||
promise: Promise<ChatEndpointRuntime>;
|
||||
version: string;
|
||||
@@ -574,7 +575,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
const discordGatewayStopTasks = new Set<Promise<void>>();
|
||||
let shuttingDown = false;
|
||||
let receiverWork: ReturnType<typeof registerChatReceiverWork> | undefined;
|
||||
const { discordGatewayNotOwnedError, isDiscordGatewayNotOwnedError, releaseDiscordGatewayLeaseRow, stopDiscordGatewayOwnership, renewDiscordGatewayOwnershipForMessageAdmission, ensureDiscordGatewayRuntimeIsCurrent, acquireDiscordGatewayOwnership, discordGatewayRuntimeIsCurrent } = createChatReceiverOwnership(db, { runtime, runtimeVersions, runtimeCallbackEndpoint, redactError, leasedChatProvider, discordGatewayOwnerships, discordGatewayStopTasks, isShuttingDown: () => shuttingDown, onStopped: id => receiverWork?.changed(id), get discordGatewayLeaseTtlMs() { return options.discordGatewayLeaseTtlMs; }, get discordGatewayLeaseRenewalIntervalMs() { return options.discordGatewayLeaseRenewalIntervalMs; }, get discordGatewayLeaseWaitMs() { return options.discordGatewayLeaseWaitMs; }, get renewDiscordGatewayLease() { return options.renewDiscordGatewayLease; } });
|
||||
const { discordGatewayNotOwnedError, isDiscordGatewayNotOwnedError, releaseDiscordGatewayLeaseRow, stopDiscordGatewayOwnership, renewDiscordGatewayOwnershipForMessageAdmission, ensureDiscordGatewayRuntimeIsCurrent, acquireDiscordGatewayOwnership, discordGatewayRuntimeIsCurrent } = createChatReceiverOwnership(db, { runtime, runtimeVersions, runtimeCallbackEndpoint, redactError, leasedChatProvider, discordGatewayOwnerships, discordGatewayStopTasks, isShuttingDown: () => shuttingDown, isRuntimeInitializing: (endpointId, context) => { const pending = runtimeInitializations.get(endpointId); return pending?.context === context && pending.discordGatewayOwned; }, onStopped: id => receiverWork?.changed(id), get discordGatewayLeaseTtlMs() { return options.discordGatewayLeaseTtlMs; }, get discordGatewayLeaseRenewalIntervalMs() { return options.discordGatewayLeaseRenewalIntervalMs; }, get discordGatewayLeaseWaitMs() { return options.discordGatewayLeaseWaitMs; }, get renewDiscordGatewayLease() { return options.renewDiscordGatewayLease; } });
|
||||
const { receiveSdkMessages } = createChatSdkMessageReceiver({
|
||||
endpointRecord, processMessage,
|
||||
ingressOnly: (provider) => provider === "imessage-photon"
|
||||
@@ -985,10 +986,12 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
}
|
||||
|
||||
const initialization: {
|
||||
context: RuntimeContext;
|
||||
discordGatewayOwned: boolean;
|
||||
promise: Promise<ChatEndpointRuntime>;
|
||||
version: string;
|
||||
} = {
|
||||
context,
|
||||
discordGatewayOwned: false,
|
||||
promise: undefined as unknown as Promise<ChatEndpointRuntime>,
|
||||
version: context.version,
|
||||
@@ -1091,6 +1094,9 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) {
|
||||
});
|
||||
context.endpointRuntime = instance;
|
||||
runtimeContexts.set(instance as object, context);
|
||||
if (discordOwnership && !(await ensureDiscordGatewayRuntimeIsCurrent(endpoint.id, context, { renewLease: true }))) {
|
||||
throw discordGatewayNotOwnedError();
|
||||
}
|
||||
await instance.initialize();
|
||||
const latest = await endpointRecord(endpoint.id);
|
||||
const stillRegistered = runtime.get(endpoint.id) === instance;
|
||||
|
||||
@@ -18,6 +18,7 @@ type OwnershipOptions = {
|
||||
discordGatewayOwnerships: Map<string, DiscordGatewayOwnership>;
|
||||
discordGatewayStopTasks: Set<Promise<void>>;
|
||||
isShuttingDown: () => boolean;
|
||||
isRuntimeInitializing: (endpointId: string, context: RuntimeContext) => boolean;
|
||||
onStopped?: (endpointId: string) => void;
|
||||
readonly discordGatewayLeaseTtlMs?: number;
|
||||
readonly discordGatewayLeaseRenewalIntervalMs?: number;
|
||||
@@ -116,19 +117,24 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
|
||||
return ownership.stopPromise;
|
||||
}
|
||||
|
||||
async function renewDiscordGatewayOwnership(
|
||||
ownership: DiscordGatewayOwnership,
|
||||
): Promise<boolean> {
|
||||
function canRenewLocalOwnership(ownership: DiscordGatewayOwnership): boolean {
|
||||
if (
|
||||
options.isShuttingDown() ||
|
||||
ownership.stopping ||
|
||||
discordGatewayOwnerships.get(ownership.endpointId) !== ownership ||
|
||||
ownership.context.discordGatewayOwned !== true ||
|
||||
ownership.context.endpointRuntime === undefined ||
|
||||
runtime.get(ownership.endpointId) !== ownership.context.endpointRuntime
|
||||
ownership.context.discordGatewayOwned !== true
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
return ownership.context.endpointRuntime === undefined
|
||||
? options.isRuntimeInitializing(ownership.endpointId, ownership.context)
|
||||
: runtime.get(ownership.endpointId) === ownership.context.endpointRuntime;
|
||||
}
|
||||
|
||||
async function renewDiscordGatewayOwnership(
|
||||
ownership: DiscordGatewayOwnership,
|
||||
): Promise<boolean> {
|
||||
if (!canRenewLocalOwnership(ownership)) return false;
|
||||
const proposedExpiresAt = new Date(
|
||||
Date.now() +
|
||||
(options.discordGatewayLeaseTtlMs ?? DISCORD_GATEWAY_LEASE_TTL_MS),
|
||||
@@ -158,7 +164,8 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
|
||||
},
|
||||
["verifying", "active", "attention"],
|
||||
);
|
||||
if (!current || !leasedChatProvider(current.provider)) return false;
|
||||
if (!current || current.companyId !== ownership.companyId || !leasedChatProvider(current.provider)
|
||||
|| !canRenewLocalOwnership(ownership)) return false;
|
||||
const expiresAt = new Date(
|
||||
now.getTime() +
|
||||
(options.discordGatewayLeaseTtlMs ?? DISCORD_GATEWAY_LEASE_TTL_MS),
|
||||
@@ -265,6 +272,7 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
|
||||
async function ensureDiscordGatewayRuntimeIsCurrent(
|
||||
endpointId: string,
|
||||
context: RuntimeContext,
|
||||
verification?: { renewLease: true },
|
||||
): Promise<boolean> {
|
||||
const ownership = discordGatewayOwnerships.get(endpointId);
|
||||
if (
|
||||
@@ -277,7 +285,12 @@ export function createChatReceiverOwnership(db: Db, options: OwnershipOptions) {
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (ownership.expiresAt.getTime() <= Date.now()) {
|
||||
if (verification?.renewLease) {
|
||||
if (ownership.renewal) await ownership.renewal;
|
||||
if (!canRenewLocalOwnership(ownership)) return false;
|
||||
await refreshDiscordGatewayOwnership(ownership);
|
||||
if (!canRenewLocalOwnership(ownership)) return false;
|
||||
} else if (ownership.expiresAt.getTime() <= Date.now()) {
|
||||
// A host can resume after the local TTL while its durable token is still
|
||||
// authoritative. Reclaim that exact token before accepting buffered
|
||||
// callbacks. A standby takeover changes the token, so the CAS fails and
|
||||
|
||||
Reference in new issue
Block a user