From fb9995231218b672992b382941a16c24b2a66a5c Mon Sep 17 00:00:00 2001 From: Patrick Erichsen Date: Fri, 31 Jul 2026 14:33:49 -0700 Subject: [PATCH] fix: reconcile native trending preflight timeouts (#3358) --- convex/httpApiV1/skillsShCatalogV1.test.ts | 38 +++++ convex/httpApiV1/skillsShCatalogV1.ts | 19 ++- scripts/skills-sh-catalog/sync.test.ts | 167 ++++++++++++++++++++- scripts/skills-sh-catalog/sync.ts | 51 ++++++- 4 files changed, 264 insertions(+), 11 deletions(-) diff --git a/convex/httpApiV1/skillsShCatalogV1.test.ts b/convex/httpApiV1/skillsShCatalogV1.test.ts index e437aa9e..b2cf4615 100644 --- a/convex/httpApiV1/skillsShCatalogV1.test.ts +++ b/convex/httpApiV1/skillsShCatalogV1.test.ts @@ -76,6 +76,44 @@ function artifact(externalId: string, content: string) { } describe("skills.sh catalog Test HTTP API", () => { + it("includes native Trending readiness in mirror status", async () => { + vi.stubEnv("CLAWHUB_ENV", "production"); + vi.stubEnv("CLAWHUB_DEPLOYMENT_NAME", "wry-manatee-359"); + vi.stubEnv("CLAWHUB_SKILLS_SH_ROLLOUT_MODE", "production"); + vi.mocked(verifyGitHubActionsSkillsShSyncJwt).mockResolvedValue({ + actor: "github-actions[bot]", + eventName: "schedule", + runId: "603", + runAttempt: "1", + sha: "a".repeat(40), + } as never); + const nativeTrending = { + status: "ready", + snapshotId: "skills-native-ready", + sourceCounts: { clawhubTrending: 10, clawhubRising: 5, skillsShTrending: 0 }, + }; + const runQuery = vi + .fn() + .mockResolvedValueOnce({ runs: [], control: null }) + .mockResolvedValueOnce(nativeTrending); + + const response = await skillsShCatalogTestV1Handler( + { runQuery } as never, + new Request("https://wry-manatee-359.convex.site/api/v1/operator/skills-sh/mirror", { + method: "POST", + headers: { Authorization: "Bearer github-oidc-token" }, + body: JSON.stringify({ operation: "mirror-status" }), + }), + ); + + await expect(response.json()).resolves.toEqual({ + runs: [], + control: null, + nativeTrending, + }); + expect(runQuery).toHaveBeenCalledTimes(2); + }); + it("accepts only the exact GitHub Actions production sync identity", async () => { const workflowSha = "a".repeat(40); vi.stubEnv("CLAWHUB_ENV", "production"); diff --git a/convex/httpApiV1/skillsShCatalogV1.ts b/convex/httpApiV1/skillsShCatalogV1.ts index ea5f60ba..11aea684 100644 --- a/convex/httpApiV1/skillsShCatalogV1.ts +++ b/convex/httpApiV1/skillsShCatalogV1.ts @@ -26,6 +26,9 @@ import { import { json, requireAdminOrResponse, requireApiTokenUserOrResponse, text } from "./shared"; const internalRefs = internal as unknown as { + canonicalTrending: { + getReadyNativeSnapshotInternal: unknown; + }; githubSkillSources: { getSkillsShAliasTargetInternal: unknown; }; @@ -666,11 +669,17 @@ export async function skillsShCatalogTestV1Handler(ctx: ActionCtx, request: Requ return text("Not found", 404, rate.headers); } if (operation === "mirror-status") { - return json( - await runQueryRef(ctx, internalRefs.skillsShMirror.getStatusInternal, {}), - 200, - rate.headers, - ); + const [status, nativeTrending] = await Promise.all([ + runQueryRef>( + ctx, + internalRefs.skillsShMirror.getStatusInternal, + {}, + ), + runQueryRef(ctx, internalRefs.canonicalTrending.getReadyNativeSnapshotInternal, { + now: Date.now(), + }), + ]); + return json({ ...status, nativeTrending }, 200, rate.headers); } if (operation === "mirror-isolation") { return json( diff --git a/scripts/skills-sh-catalog/sync.test.ts b/scripts/skills-sh-catalog/sync.test.ts index 1093854b..b67ccf87 100644 --- a/scripts/skills-sh-catalog/sync.test.ts +++ b/scripts/skills-sh-catalog/sync.test.ts @@ -160,6 +160,171 @@ describe("skills.sh synchronization runner", () => { ]); }); + it("waits for a timed-out native preflight to finish without starting it twice", async () => { + const operations: string[] = []; + const sleep = vi.fn(async () => undefined); + let statusCalls = 0; + const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => { + const body = JSON.parse(String(init.body)) as Record; + operations.push(String(body.operation)); + if (body.operation === "status") { + statusCalls += 1; + if (statusCalls === 1) { + return response({ runs: [], invariants: { publicVisible: false } }); + } + if (statusCalls === 2) { + return response({ + control: { + activationLockToken: "skills-sh-native-trending:in-flight", + activationLockedAt: 1_722_345_678_000, + reason: "scheduled proof native-only preflight", + }, + nativeTrending: null, + runs: [], + invariants: { publicVisible: false }, + }); + } + if (statusCalls === 3) { + return response({ + control: {}, + nativeTrending: { + status: "ready", + snapshotId: "skills-native-after-timeout", + sourceCounts: { clawhubTrending: 1, clawhubRising: 1, skillsShTrending: 0 }, + }, + runs: [], + invariants: { publicVisible: false }, + }); + } + return response({ runs: [], invariants: { publicVisible: true } }); + } + if (body.operation === "prepare-native-trending") { + throw new DOMException("The operation timed out.", "TimeoutError"); + } + if (body.operation === "configure") return response({ ok: true }); + if (body.operation === "start") return response(completedRun("leaderboard")); + if (body.operation === "start-trending") return response(completedRun("trending")); + if (body.operation === "verify-activate") { + return response({ ok: true, activated: true }); + } + throw new Error(`unexpected operation ${String(body.operation)}`); + }); + + await expect( + runSkillsShSync({ + targetUrl: "https://clawhub.ai/ops/skills-sh/mirror", + authorization: "github-oidc", + reason: "scheduled proof", + fetchImpl, + sleep, + }), + ).resolves.toMatchObject({ + ok: true, + nativeBefore: { + nativeTrending: { + status: "ready", + snapshotId: "skills-native-after-timeout", + sourceCounts: { skillsShTrending: 0 }, + }, + reconciledAfterTimeout: true, + }, + }); + expect(sleep).toHaveBeenCalledExactlyOnceWith(5_000); + expect(operations).toEqual([ + "status", + "configure", + "prepare-native-trending", + "status", + "status", + "start", + "start-trending", + "verify-activate", + "configure", + "status", + ]); + }); + + it("fails closed when a timed-out native preflight releases without a ready snapshot", async () => { + const operations: string[] = []; + const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => { + const body = JSON.parse(String(init.body)) as Record; + operations.push(String(body.operation)); + if (body.operation === "status") { + return operations.length === 1 + ? response({ runs: [], invariants: { publicVisible: false } }) + : response({ + control: {}, + nativeTrending: null, + runs: [], + invariants: { publicVisible: false }, + }); + } + if (body.operation === "prepare-native-trending") { + throw new DOMException("The operation timed out.", "TimeoutError"); + } + if (body.operation === "configure") return response({ ok: true }); + throw new Error(`unexpected operation ${String(body.operation)}`); + }); + + await expect( + runSkillsShSync({ + targetUrl: "https://clawhub.ai/ops/skills-sh/mirror", + authorization: "github-oidc", + reason: "scheduled proof", + fetchImpl, + }), + ).rejects.toThrow("native-only Trending preflight finished without a ready snapshot"); + expect(operations).toEqual([ + "status", + "configure", + "prepare-native-trending", + "status", + "configure", + ]); + }); + + it("fails closed when a timed-out native preflight is replaced by another lock", async () => { + const operations: string[] = []; + const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => { + const body = JSON.parse(String(init.body)) as Record; + operations.push(String(body.operation)); + if (body.operation === "status") { + return operations.length === 1 + ? response({ runs: [], invariants: { publicVisible: false } }) + : response({ + control: { + activationLockToken: "skills-sh-activation:different-operation", + reason: "another activation", + }, + nativeTrending: null, + runs: [], + invariants: { publicVisible: false }, + }); + } + if (body.operation === "prepare-native-trending") { + throw new DOMException("The operation timed out.", "TimeoutError"); + } + if (body.operation === "configure") return response({ ok: true }); + throw new Error(`unexpected operation ${String(body.operation)}`); + }); + + await expect( + runSkillsShSync({ + targetUrl: "https://clawhub.ai/ops/skills-sh/mirror", + authorization: "github-oidc", + reason: "scheduled proof", + fetchImpl, + }), + ).rejects.toThrow("timed-out native Trending preflight is bound to a different lock"); + expect(operations).toEqual([ + "status", + "configure", + "prepare-native-trending", + "status", + "configure", + ]); + }); + it("resumes an interrupted durable run before starting the next source view", async () => { const operations: string[] = []; const fetchImpl = vi.fn(async (_url: string, init: RequestInit) => { @@ -567,7 +732,7 @@ describe("skills.sh synchronization runner", () => { sleep, }), ).rejects.toThrow("did not produce an exact durable activation receipt"); - expect(sleep).toHaveBeenCalledTimes(131); + expect(sleep).toHaveBeenCalledTimes(359); }); it("preserves the last verified public lane on a transient sync failure", async () => { diff --git a/scripts/skills-sh-catalog/sync.ts b/scripts/skills-sh-catalog/sync.ts index 1553452b..27dac3d9 100644 --- a/scripts/skills-sh-catalog/sync.ts +++ b/scripts/skills-sh-catalog/sync.ts @@ -14,8 +14,9 @@ const MAX_STEPS = 2_000; const MAX_RATE_LIMIT_RETRIES = 30; const MAX_RATE_LIMIT_WAIT_MS = 30 * 60 * 1_000; const MAX_TRANSPORT_TIMEOUTS = 3; +const MAX_NATIVE_TRENDING_RECONCILE_POLLS = 360; const ACTIVATION_RECONCILE_POLL_MS = 5_000; -const MAX_ACTIVATION_RECONCILE_POLLS = 132; +const MAX_ACTIVATION_RECONCILE_POLLS = 360; type MirrorRun = Record; type SyncFetch = (input: string, init: RequestInit) => Promise; @@ -308,6 +309,40 @@ export async function runSkillsShSync(options: { ); }; + const reconcileTimedOutNativePreparation = async ( + preflightReason: string, + ): Promise => { + for (let poll = 0; poll < MAX_NATIVE_TRENDING_RECONCILE_POLLS; poll += 1) { + const status = await call({ operation: "status" }); + const control = optionalRecord(status.control); + const lockToken = control?.activationLockToken; + if (typeof lockToken === "string") { + if ( + !lockToken.startsWith("skills-sh-native-trending:") || + control?.reason !== preflightReason + ) { + throw new Error("timed-out native Trending preflight is bound to a different lock"); + } + if (poll + 1 < MAX_NATIVE_TRENDING_RECONCILE_POLLS) { + await sleep(ACTIVATION_RECONCILE_POLL_MS); + continue; + } + break; + } + const nativeTrending = optionalRecord(status.nativeTrending); + const sourceCounts = optionalRecord(nativeTrending?.sourceCounts); + if (nativeTrending?.status !== "ready" || sourceCounts?.skillsShTrending !== 0) { + throw new Error("native-only Trending preflight finished without a ready snapshot"); + } + return { + ok: true, + nativeTrending, + reconciledAfterTimeout: true, + }; + } + throw new Error("timed-out native Trending preflight did not release its exact durable lock"); + }; + const startedAt = Date.now(); const before = await call({ operation: "status" }); const publicVisible = (before.invariants as Record | undefined)?.publicVisible; @@ -320,12 +355,18 @@ export async function runSkillsShSync(options: { let nativeBefore: Record | null = null; await call({ operation: "configure", enabled: true, reason: options.reason }); try { - nativeBefore = publicVisible - ? null - : await call({ + if (!publicVisible) { + const preflightReason = `${options.reason} native-only preflight`; + try { + nativeBefore = await call({ operation: "prepare-native-trending", - reason: `${options.reason} native-only preflight`, + reason: preflightReason, }); + } catch (error) { + if (!isTransportTimeout(error)) throw error; + nativeBefore = await reconcileTimedOutNativePreparation(preflightReason); + } + } if (nativeBefore) { const nativeTrending = nativeBefore.nativeTrending as Record | undefined; const sourceCounts = nativeTrending?.sourceCounts as Record | undefined;