mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 14:16:39 +01:00
fix: preserve GitHub redeliveries stopped before transport
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
936e132c01
commit
491a7e1eef
4 files changed
+155
-25
No files matched your search
@@ -2026,3 +2026,19 @@ acceptance and remote gates remain required. See
|
||||
Earlier #17490/#17572 CI failures, #17576/#17580 review findings, the entire
|
||||
remaining original list, and final reconciliation with master stay open. The
|
||||
current inspected master is `5a28f7acf8`; no stack PR has been merged.
|
||||
|
||||
|
||||
## Slice 38 review repair and corrected acceptance
|
||||
|
||||
Initial #17666 at `936e132c01` has 52 passing checks and a valid Greptile P1:
|
||||
a receipt committed immediately before drain/stop could suppress a POST that
|
||||
never started. Four regressions reproduced the loss. The owner now restores a
|
||||
known-unsent claim with an exact attempt predicate while retaining uncertainty
|
||||
after transport starts. Fifty-nine selected tests pass; six stop-boundary cases
|
||||
also cover preservation of a newer receipt. Final verification is running.
|
||||
|
||||
The initial broad command passed 1,586 tests in two files but misspelled the
|
||||
standalone publication filename. The corrected three-file command is running
|
||||
on a fresh database; the earlier result is not full publication acceptance.
|
||||
Earlier stack CI/review issues, all remaining migrations, and master reconciliation
|
||||
remain outstanding. No stack PR is merged or claimed ready as a whole.
|
||||
@@ -131,6 +131,41 @@ process. No permanent URL-discovery poll is introduced.
|
||||
acceptance and remote CI/review remain required. This is not a full-stack
|
||||
readiness claim.
|
||||
|
||||
## Review repair: a claimed redelivery that never reached transport
|
||||
|
||||
Greptile identified a valid P1 at initial head `936e132c01`: drain/shutdown
|
||||
could start after the recovery receipt committed and before the POST began.
|
||||
The guard correctly prevented HTTP, but the catch marked the receipt uncertain,
|
||||
and the next scan refused that same failed delivery. Four real-PostgreSQL cases
|
||||
reproduced the lost callback for drain/shutdown with both a first claim and a
|
||||
previous receipt (`/tmp/sleep38-unsent-regression.log`, four failures).
|
||||
|
||||
The owner now distinguishes transport admission from a persisted claim. If the
|
||||
request did not start, it restores the prior receipt or removes the new receipt
|
||||
inside a short notifying transaction. If transport may have started, it retains
|
||||
`delivery_unknown`. Settlement and restoration use the exact attempt identity
|
||||
(id, processing status, count, provider delivery, and requested timestamp), so
|
||||
an older attempt cannot overwrite a successor. Lease failure before HTTP follows
|
||||
the same known-unsent path. Neither database transactions nor the provider POST
|
||||
are replayed automatically. A crash between commit and transport, or failure to
|
||||
persist restoration, still requires conservative durable uncertainty handling.
|
||||
|
||||
- Server typecheck and 59 selected integration/publication/startup cases passed
|
||||
after the repair (`/tmp/sleep38-unsent-types.log`,
|
||||
`/tmp/sleep38-unsent-fixed.log`, 28.91 seconds).
|
||||
- Two additional cases replace the receipt timestamp while preserving its
|
||||
processing status and attempt count. All six stop-before-transport cases pass;
|
||||
restoration leaves the newer receipt intact and restart does not replay it
|
||||
(`/tmp/sleep38-unsent-cas.log`, 10.27 seconds).
|
||||
- The initial full command passed 1,586 tests across two files in 571.36 seconds
|
||||
(`/tmp/sleep38-full-chat.log`). It misspelled the standalone publication suite
|
||||
filename, so it did not run that suite and is not three-suite acceptance.
|
||||
The corrected complete command includes `chat-interaction-publications.test.ts`.
|
||||
- Fresh final workspace typecheck/build and corrected three-suite acceptance on
|
||||
`paperclip_sleep38_final_20261025_1837_test` are running. Initial remote CI has
|
||||
52 passing checks and the valid failing Greptile review; repaired-head gates
|
||||
remain required.
|
||||
|
||||
## Remaining boundaries
|
||||
|
||||
The idle-safety proof still blocks unsupported active channel obligations. A
|
||||
|
||||
@@ -15641,6 +15641,70 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
|
||||
} finally { await coordinator.stop(); vi.useRealTimers(); }
|
||||
});
|
||||
|
||||
it.each([
|
||||
["drain", false, false], ["shutdown", false, false],
|
||||
["drain", true, false], ["shutdown", true, false],
|
||||
["drain", false, true], ["shutdown", true, true],
|
||||
] as const)("preserves a GitHub redelivery stopped before transport (%s, prior receipt: %s, superseded: %s)", async (mode, priorReceipt, superseded) => {
|
||||
const f = await githubPreIngressRecoveryFixture();
|
||||
await f.makeScanDue();
|
||||
if (priorReceipt) {
|
||||
await f.service.processFailedGitHubWebhookDeliveries(5, f.endpoint.id);
|
||||
const sent = (await f.receipt())!;
|
||||
await db.update(chatActions).set({ result: { ...sent.result, retryAt: new Date(0).toISOString(),
|
||||
requestedAt: new Date(Date.now() - 60_000).toISOString() } }).where(eq(chatActions.id, sent.id));
|
||||
f.setDeliveries([{ ...f.original, id: "9007199254740994", redelivery: true,
|
||||
delivered_at: new Date(Date.now() - 15_000).toISOString() }, f.original]);
|
||||
await f.makeScanDue();
|
||||
}
|
||||
const previous = await f.receipt();
|
||||
let successor: Awaited<ReturnType<typeof f.receipt>>;
|
||||
f.posts.length = 0;
|
||||
let canRun = true, stoppedAfterClaim = false;
|
||||
let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, canReconcile: () => true,
|
||||
onError: () => undefined });
|
||||
let stopping: Promise<void> | undefined;
|
||||
const transaction = db.transaction.bind(db);
|
||||
const transactions = vi.spyOn(db, "transaction").mockImplementation((async (callback, options) => {
|
||||
const result = await transaction(callback, options);
|
||||
if (typeof result === "string" && !stoppedAfterClaim) {
|
||||
const receipt = await f.receipt();
|
||||
if (receipt?.id === result && receipt.status === "processing") {
|
||||
if (superseded) {
|
||||
await db.update(chatActions).set({ result: { ...receipt.result,
|
||||
requestedAt: new Date(Date.now() + 1000).toISOString() } }).where(eq(chatActions.id, receipt.id));
|
||||
successor = await f.receipt();
|
||||
}
|
||||
stoppedAfterClaim = true; canRun = false;
|
||||
if (mode === "shutdown") stopping = coordinator.stop();
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}) as Db["transaction"]);
|
||||
try {
|
||||
await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => canRun).ready;
|
||||
await stopping;
|
||||
expect(stoppedAfterClaim).toBe(true); expect(f.posts).toEqual([]);
|
||||
if (superseded) expect(await f.receipt()).toEqual(successor);
|
||||
else if (previous) expect(await f.receipt()).toMatchObject({ status: previous.status,
|
||||
result: previous.result, payload: previous.payload });
|
||||
else expect(await f.receipt()).toBeUndefined();
|
||||
transactions.mockRestore();
|
||||
await coordinator.stop();
|
||||
await f.makeScanDue();
|
||||
canRun = true;
|
||||
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
|
||||
await f.service.registerGitHubWebhookRecoveryWork(coordinator, () => true).ready;
|
||||
if (superseded) {
|
||||
expect(f.posts).toEqual([]);
|
||||
expect(await f.receipt()).toEqual(successor);
|
||||
} else {
|
||||
expect(f.posts).toHaveLength(1);
|
||||
expect(await f.receipt()).toMatchObject({ status: "processed", result: { attempts: priorReceipt ? 2 : 1 } });
|
||||
}
|
||||
} finally { transactions.mockRestore(); await coordinator.stop(); await stopping; }
|
||||
});
|
||||
|
||||
it("joins held GitHub history at shutdown without starting another HTTP request", async () => {
|
||||
const f = await githubPreIngressRecoveryFixture(); await f.makeScanDue();
|
||||
let release!: () => void, entered!: () => void;
|
||||
|
||||
@@ -1142,8 +1142,11 @@ export function createGitHubWebhookIngress(db: Db, options: GitHubWebhookIngress
|
||||
) {
|
||||
const admitted = () => !isShuttingDown() && canRun() && !signal?.aborted;
|
||||
if (!admitted() || !getWebhookPublicBaseUrl()?.startsWith("https://")) return;
|
||||
const recoveryFetch: typeof fetch = (input, init) => {
|
||||
const assertAdmitted = () => {
|
||||
if (!admitted()) throw signal?.reason ?? new Error("GitHub webhook recovery stopped");
|
||||
};
|
||||
const recoveryFetch: typeof fetch = (input, init) => {
|
||||
assertAdmitted();
|
||||
return fetchImpl(input, init);
|
||||
};
|
||||
const now = new Date();
|
||||
@@ -1403,6 +1406,7 @@ export function createGitHubWebhookIngress(db: Db, options: GitHubWebhookIngress
|
||||
currentConfig.insecureSsl !== "0"
|
||||
)
|
||||
return;
|
||||
const requestedAt = new Date().toISOString();
|
||||
const claimed = await db.transaction(async (tx) => {
|
||||
await lease.assertOwned(tx);
|
||||
const current = await runtimeCallbackEndpoint(
|
||||
@@ -1460,7 +1464,7 @@ export function createGitHubWebhookIngress(db: Db, options: GitHubWebhookIngress
|
||||
const result = {
|
||||
attempts: count + 1,
|
||||
latestAttemptId: latest.id,
|
||||
requestedAt: new Date().toISOString(),
|
||||
requestedAt,
|
||||
retryAt: new Date(
|
||||
Date.now() + 60_000 * 5 ** count,
|
||||
).toISOString(),
|
||||
@@ -1500,38 +1504,49 @@ export function createGitHubWebhookIngress(db: Db, options: GitHubWebhookIngress
|
||||
return changes[0]?.id ?? null;
|
||||
});
|
||||
if (!claimed) return;
|
||||
requested += 1;
|
||||
// Commit the immutable denial fence before transport. Even an
|
||||
// ambiguous POST is not reattempted without a distinct failed
|
||||
// provider delivery. A stale scanner cannot overwrite a successor.
|
||||
await lease.assertOwned();
|
||||
const ownsAttempt = and(
|
||||
eq(chatActions.id, claimed),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${count + 1}`,
|
||||
sql`${chatActions.result}->>'latestAttemptId' = ${latest.id}`,
|
||||
sql`${chatActions.result}->>'requestedAt' = ${requestedAt}`,
|
||||
);
|
||||
let transportStarted = false;
|
||||
try {
|
||||
await lease.assertOwned();
|
||||
await requestGitHubAppWebhookRedelivery({
|
||||
fetch: recoveryFetch,
|
||||
fetch: (input, init) => {
|
||||
assertAdmitted();
|
||||
transportStarted = true;
|
||||
requested += 1;
|
||||
return fetchImpl(input, init);
|
||||
},
|
||||
appToken,
|
||||
deliveryId: original.id,
|
||||
});
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({ status: "processed", updatedAt: new Date() })
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, claimed),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${count + 1}`,
|
||||
),
|
||||
);
|
||||
.where(ownsAttempt);
|
||||
} catch (error) {
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({ status: "delivery_unknown", updatedAt: new Date() })
|
||||
.where(
|
||||
and(
|
||||
eq(chatActions.id, claimed),
|
||||
eq(chatActions.status, "processing"),
|
||||
sql`(${chatActions.result}->>'attempts')::int = ${count + 1}`,
|
||||
),
|
||||
);
|
||||
if (transportStarted) {
|
||||
await db
|
||||
.update(chatActions)
|
||||
.set({ status: "delivery_unknown", updatedAt: new Date() })
|
||||
.where(ownsAttempt);
|
||||
} else {
|
||||
await db.transaction(async (tx) => {
|
||||
await notifyGitHubWebhookRecovery(tx, { kind: "endpoint", id: endpoint.id });
|
||||
if (receipt) {
|
||||
await tx
|
||||
.update(chatActions)
|
||||
.set({ status: receipt.status, result: receipt.result, updatedAt: receipt.updatedAt })
|
||||
.where(ownsAttempt);
|
||||
} else {
|
||||
await tx.delete(chatActions).where(ownsAttempt);
|
||||
}
|
||||
});
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
|
||||
Reference in new issue
Block a user