diff --git a/convex/publishAttempts.runtime.test.ts b/convex/publishAttempts.runtime.test.ts new file mode 100644 index 00000000..43389ef4 --- /dev/null +++ b/convex/publishAttempts.runtime.test.ts @@ -0,0 +1,146 @@ +/// +/* @vitest-environment edge-runtime */ +import { convexTest } from "convex-test"; +import { describe, expect, it } from "vitest"; +import { internal } from "./_generated/api"; +import schema from "./schema"; + +const modules = import.meta.glob("./**/*.ts"); + +describe("publish attempt orphan recovery", () => { + it("terminalizes a pending attempt after its staged version is deleted", async () => { + const t = convexTest(schema, modules); + const ids = await t.run(async (ctx) => { + const userId = await ctx.db.insert("users", {}); + const skillId = await ctx.db.insert("skills", { + slug: "orphan-runtime", + displayName: "Orphan Runtime", + ownerUserId: userId, + forkOf: undefined, + tags: {}, + stats: { comments: 0, downloads: 0, stars: 0, versions: 0 }, + createdAt: 1, + updatedAt: 1, + }); + const versionId = await ctx.db.insert("skillVersions", { + skillId, + version: "1.0.0", + publicationStatus: "pending", + changelog: "", + files: [], + parsed: { frontmatter: {} }, + createdBy: userId, + createdAt: 1, + }); + const attemptId = await ctx.db.insert("publishAttempts", { + kind: "skill", + status: "pending_checks", + userId, + skillId, + skillVersionId: versionId, + slug: "orphan-runtime", + displayName: "Orphan Runtime", + version: "1.0.0", + idempotencyKey: "runtime-orphan", + artifactFingerprint: "fingerprint", + files: [], + checks: { + trufflehog: { status: "pending" }, + clawscan: { status: "pending" }, + }, + createdAt: 1, + updatedAt: 1, + expiresAt: Date.now() + 60_000, + }); + await ctx.db.delete(versionId); + return { attemptId }; + }); + + await expect( + t.mutation(internal.publishAttempts.claimPendingPublishAttemptChecksInternal, { + attemptId: ids.attemptId, + claimId: "runtime-claim", + }), + ).resolves.toBeNull(); + + const attempt = await t.run(async (ctx) => ctx.db.get(ids.attemptId)); + expect(attempt).toMatchObject({ + status: "failed", + checkClaimLastError: "Pending skill version not found.", + failedAt: expect.any(Number), + }); + }); + + it("terminalizes a ready attempt after its staged release is soft-deleted", async () => { + const t = convexTest(schema, modules); + const ids = await t.run(async (ctx) => { + const userId = await ctx.db.insert("users", {}); + const packageId = await ctx.db.insert("packages", { + name: "@demo/orphan-runtime", + normalizedName: "@demo/orphan-runtime", + displayName: "Orphan Runtime", + ownerUserId: userId, + family: "code-plugin", + channel: "community", + isOfficial: false, + tags: {}, + compatibility: {}, + verification: { tier: "structural", scope: "artifact-only", scanStatus: "pending" }, + scanStatus: "pending", + stats: { downloads: 0, installs: 0, stars: 0, versions: 0 }, + createdAt: 1, + updatedAt: 1, + }); + const releaseId = await ctx.db.insert("packageReleases", { + packageId, + version: "1.0.0", + publicationStatus: "pending", + changelog: "", + distTags: [], + files: [], + integritySha256: "fingerprint", + compatibility: {}, + verification: { tier: "structural", scope: "artifact-only", scanStatus: "pending" }, + createdBy: userId, + publishActor: { kind: "user", userId }, + createdAt: 1, + softDeletedAt: 2, + }); + const attemptId = await ctx.db.insert("publishAttempts", { + kind: "package", + status: "ready_to_finalize", + userId, + packageId, + packageReleaseId: releaseId, + slug: "@demo/orphan-runtime", + displayName: "Orphan Runtime", + version: "1.0.0", + idempotencyKey: "runtime-orphan-package", + artifactFingerprint: "fingerprint", + files: [], + checks: { + trufflehog: { status: "clean" }, + clawscan: { status: "clean" }, + }, + createdAt: 1, + updatedAt: 1, + expiresAt: Date.now() + 60_000, + }); + return { attemptId }; + }); + + await expect( + t.mutation(internal.publishAttempts.claimReadyPublishAttemptFinalizationRetryInternal, { + attemptId: ids.attemptId, + claimId: "runtime-finalize", + }), + ).resolves.toBeNull(); + + const attempt = await t.run(async (ctx) => ctx.db.get(ids.attemptId)); + expect(attempt).toMatchObject({ + status: "failed", + finalizationLastError: "Pending package release not found", + failedAt: expect.any(Number), + }); + }); +}); diff --git a/convex/publishAttempts.test.ts b/convex/publishAttempts.test.ts index 66addcbb..d137a575 100644 --- a/convex/publishAttempts.test.ts +++ b/convex/publishAttempts.test.ts @@ -165,6 +165,71 @@ describe("publishAttempts", () => { ); }); + it("terminalizes orphaned pending attempts and claims healthy work behind them", async () => { + const orphan = { + _id: "publishAttempts:orphan", + kind: "skill", + status: "pending_checks", + userId: "users:publisher", + skillVersionId: "skillVersions:deleted", + slug: "deleted-skill", + displayName: "Deleted Skill", + version: "1.0.0", + artifactFingerprint: "fingerprint", + files: [{ path: "SKILL.md", storageId: "_storage:skill", size: 10, sha256: "sha" }], + createdAt: Date.now(), + }; + const healthy = { + ...orphan, + _id: "publishAttempts:healthy", + skillVersionId: "skillVersions:healthy", + slug: "healthy-skill", + }; + const ctx = { + db: { + delete: vi.fn(), + get: vi.fn(async (id: string) => + id === "skillVersions:healthy" + ? { _id: id, fingerprint: healthy.artifactFingerprint } + : null, + ), + insert: vi.fn(), + normalizeId: vi.fn(), + patch: vi.fn(), + query: vi.fn(() => ({ + withIndex: vi.fn(() => ({ + order: vi.fn(() => ({ + take: vi.fn(async () => [orphan, healthy]), + })), + })), + })), + replace: vi.fn(), + system: {}, + }, + }; + + await expect( + claimPendingChecksHandler(ctx, { claimId: "checks:claim" }), + ).resolves.toMatchObject({ + attemptId: "publishAttempts:healthy", + slug: "healthy-skill", + }); + + expect(ctx.db.patch).toHaveBeenCalledWith( + "publishAttempts:orphan", + expect.objectContaining({ + status: "failed", + checkClaimId: undefined, + checkClaimLastError: "Pending skill version not found.", + failedAt: expect.any(Number), + }), + ); + expect(ctx.db.patch).toHaveBeenCalledWith( + "publishAttempts:healthy", + expect.objectContaining({ checkClaimId: "checks:claim" }), + ); + }); + it("reuses a completed ClawScan verdict only for the exact staged artifact", async () => { const attempt = { _id: "publishAttempts:reusable", @@ -391,6 +456,92 @@ describe("publishAttempts", () => { expect(ctx.db.patch).not.toHaveBeenCalled(); }); + it("terminalizes orphaned ready attempts and claims healthy work behind them", async () => { + const orphan = { + _id: "publishAttempts:orphan-package", + kind: "package", + status: "ready_to_finalize", + packageReleaseId: "packageReleases:deleted", + slug: "@demo/deleted", + version: "1.0.0", + createdAt: Date.now(), + }; + const healthy = { + ...orphan, + _id: "publishAttempts:healthy-package", + packageReleaseId: "packageReleases:healthy", + slug: "@demo/healthy", + }; + const ctx = { + db: { + delete: vi.fn(), + get: vi.fn(async (id: string) => + id === "packageReleases:deleted" ? { _id: id, softDeletedAt: Date.now() } : { _id: id }, + ), + insert: vi.fn(), + normalizeId: vi.fn(), + patch: vi.fn(), + query: vi.fn(() => ({ + withIndex: vi.fn(() => ({ + order: vi.fn(() => ({ + take: vi.fn(async () => [orphan, healthy]), + })), + })), + })), + replace: vi.fn(), + system: {}, + }, + }; + + await expect( + claimReadyFinalizationHandler(ctx, { claimId: "finalize:claim" }), + ).resolves.toMatchObject({ + attemptId: "publishAttempts:healthy-package", + slug: "@demo/healthy", + }); + + expect(ctx.db.patch).toHaveBeenCalledWith( + "publishAttempts:orphan-package", + expect.objectContaining({ + status: "failed", + finalizationClaimId: undefined, + finalizationLastError: "Pending package release not found", + failedAt: expect.any(Number), + }), + ); + expect(ctx.db.patch).toHaveBeenCalledWith( + "publishAttempts:healthy-package", + expect.objectContaining({ checkClaimId: "finalize:claim" }), + ); + }); + + it("treats targeted attempts terminalized by the ready queue as drained", async () => { + const ctx = { + db: { + delete: vi.fn(), + get: vi.fn(async () => ({ + _id: "publishAttempts:orphan-package", + kind: "package", + status: "failed", + })), + insert: vi.fn(), + normalizeId: vi.fn(), + patch: vi.fn(), + query: vi.fn(), + replace: vi.fn(), + system: {}, + }, + }; + + await expect( + claimPendingChecksHandler(ctx, { + attemptId: "publishAttempts:orphan-package", + claimId: "finalize:claim", + }), + ).resolves.toBeNull(); + expect(ctx.db.patch).not.toHaveBeenCalled(); + }); + it("rejects targeted ready-finalization claims with mismatched filters", async () => { const ctx = { db: { @@ -520,6 +671,74 @@ describe("publishAttempts", () => { expect(patch.checkClaimExpiresAt).toBeGreaterThan(now); }); + it("terminalizes an attempt when its staged target disappears during scanning", async () => { + const now = Date.now(); + const ctx = { + db: { + get: vi.fn(async (id: string) => + id === "publishAttempts:orphan" + ? { + _id: "publishAttempts:orphan", + kind: "skill", + status: "pending_checks", + skillVersionId: "skillVersions:deleted", + artifactFingerprint: "fingerprint", + checkClaimId: "checks:claim", + checkClaimExpiresAt: now + 60_000, + checks: { + trufflehog: { status: "pending" }, + clawscan: { status: "pending" }, + }, + } + : null, + ), + patch: vi.fn(async (id: string) => { + if (id === "skillVersions:deleted") { + throw new Error("Update on nonexistent document ID skillVersions:deleted"); + } + }), + insert: vi.fn(), + replace: vi.fn(), + delete: vi.fn(), + query: vi.fn(), + normalizeId: vi.fn(), + system: {}, + }, + storage: { + delete: vi.fn(), + }, + }; + + await expect( + completePendingChecksHandler(ctx, { + attemptId: "publishAttempts:orphan", + claimId: "checks:claim", + artifactFingerprint: "fingerprint", + trufflehog: { status: "clean" }, + clawscan: { status: "clean" }, + clawscanAnalysis: { + status: "clean", + verdict: "benign", + checkedAt: now, + }, + }), + ).resolves.toEqual({ + attemptId: "publishAttempts:orphan", + kind: "skill", + status: "failed", + }); + + expect(ctx.db.patch).toHaveBeenCalledWith( + "publishAttempts:orphan", + expect.objectContaining({ + status: "failed", + checkClaimLastError: "Pending skill version not found.", + failedAt: expect.any(Number), + }), + ); + expect(ctx.db.patch).not.toHaveBeenCalledWith("skillVersions:deleted", expect.anything()); + }); + it("terminalizes duplicate skill versions instead of retrying finalization", async () => { const ctx = { db: { @@ -619,6 +838,7 @@ describe("publishAttempts", () => { "Uncaught ConvexError: Slug redirects to an existing skill. Choose a different slug. Existing skill: /orchune/personal-finance", ], ["deleted fork sources", "Uncaught ConvexError: Upstream skill not found"], + ["deleted staged versions", "Uncaught ConvexError: Pending skill version not found."], ])("terminalizes %s instead of retrying finalization", async (_caseName, error) => { const ctx = { db: { @@ -734,6 +954,48 @@ describe("publishAttempts", () => { expect(transientCtx.db.patch.mock.calls[0]?.[1]).not.toHaveProperty("failedAt"); }); + it("terminalizes deleted package releases instead of retrying finalization", async () => { + const ctx = { + db: { + delete: vi.fn(), + get: vi.fn(async () => ({ + _id: "publishAttempts:orphan-package", + kind: "package", + status: "finalizing", + packageReleaseId: "packageReleases:deleted", + finalizationClaimId: "finalize:claim", + })), + insert: vi.fn(), + normalizeId: vi.fn(), + patch: vi.fn(), + query: vi.fn(), + replace: vi.fn(), + system: {}, + }, + }; + const error = "Uncaught ConvexError: Pending package release not found"; + + await expect( + releasePackageFinalizationHandler(ctx, { + attemptId: "publishAttempts:orphan-package", + claimId: "finalize:claim", + error, + }), + ).resolves.toEqual({ + attemptId: "publishAttempts:orphan-package", + status: "failed", + }); + + expect(ctx.db.patch).toHaveBeenCalledWith( + "publishAttempts:orphan-package", + expect.objectContaining({ + status: "failed", + finalizationLastError: error, + failedAt: expect.any(Number), + }), + ); + }); + it("clears private pending skill metadata when finalization is recorded", async () => { const now = Date.now(); const ctx = { diff --git a/convex/publishAttempts.ts b/convex/publishAttempts.ts index f1af23e1..56723f39 100644 --- a/convex/publishAttempts.ts +++ b/convex/publishAttempts.ts @@ -1,6 +1,6 @@ import { ConvexError, v } from "convex/values"; import { internal } from "./_generated/api"; -import type { Id } from "./_generated/dataModel"; +import type { Doc, Id } from "./_generated/dataModel"; import type { MutationCtx } from "./_generated/server"; import { action, internalAction, internalMutation, internalQuery } from "./functions"; import { finalizeSkillPublishAttempt } from "./lib/skillPublish"; @@ -112,7 +112,9 @@ function isTerminalFinalizationConflict(error: string | undefined) { (/Version .+ already exists\. Increment the version number and try again\./.test(error) || error.includes("Slug is used by multiple publishers. Use an owner-qualified skill URL.") || error.includes("Slug redirects to an existing skill. Choose a different slug.") || - error.includes("Upstream skill not found")) + error.includes("Upstream skill not found") || + error.includes("Pending skill version not found.") || + error.includes("Pending package release not found")) ); } @@ -142,6 +144,45 @@ function releaseFinalizationClaimPatch(error: string | undefined, now: number) { }; } +async function unavailableStagedTargetError( + ctx: Pick, + attempt: Doc<"publishAttempts">, +) { + if (attempt.kind === "skill" && attempt.skillVersionId) { + const version = await ctx.db.get(attempt.skillVersionId); + if (!version || version.softDeletedAt) return "Pending skill version not found."; + } + if (attempt.kind === "package" && attempt.packageReleaseId) { + const release = await ctx.db.get(attempt.packageReleaseId); + if (!release || release.softDeletedAt) return "Pending package release not found"; + } + return null; +} + +async function terminalizeUnavailableStagedTarget( + ctx: Pick, + attempt: Doc<"publishAttempts">, + now: number, +) { + const error = await unavailableStagedTargetError(ctx, attempt); + if (!error) return false; + const pendingChecks = attempt.status === "pending_checks"; + await ctx.db.patch(attempt._id, { + status: "failed", + checkClaimId: undefined, + checkClaimedAt: undefined, + checkClaimExpiresAt: undefined, + checkClaimLastError: pendingChecks ? error : undefined, + finalizationClaimId: undefined, + finalizationClaimedAt: undefined, + finalizationClaimExpiresAt: undefined, + finalizationLastError: pendingChecks ? undefined : error, + failedAt: now, + updatedAt: now, + }); + return true; +} + export const createSkillPublishAttemptInternal = internalMutation({ args: { userId: v.id("users"), @@ -614,6 +655,10 @@ export const completePendingPublishAttemptChecksInternal = internalMutation({ return { attemptId: attempt._id, kind: attempt.kind, status: "blocked" as const }; } + if (await terminalizeUnavailableStagedTarget(ctx, attempt, now)) { + return { attemptId: attempt._id, kind: attempt.kind, status: "failed" as const }; + } + if (args.trufflehog.status === "failed" || args.clawscan.status === "failed") { await ctx.db.patch(attempt._id, { status: "pending_checks", @@ -672,93 +717,107 @@ export const claimPendingPublishAttemptChecksInternal = internalMutation({ }, handler: async (ctx, args) => { const now = Date.now(); - const attempt = args.attemptId - ? await ctx.db.get(args.attemptId) - : ( - await ctx.db - .query("publishAttempts") - .withIndex("by_status_check_claim_expires_at_created", (q) => - q.eq("status", "pending_checks"), - ) - .order("asc") - .take(25) - ).find((candidate) => { - if ((candidate.checkClaimExpiresAt ?? 0) > now) return false; - if (args.kind && candidate.kind !== args.kind) return false; - if (args.slug && candidate.slug !== args.slug) return false; - if (args.version && candidate.version !== args.version) return false; - return true; - }); + const targetedAttempt = args.attemptId ? await ctx.db.get(args.attemptId) : null; + const candidates = args.attemptId + ? targetedAttempt + ? [targetedAttempt] + : [] + : await ctx.db + .query("publishAttempts") + .withIndex("by_status_check_claim_expires_at_created", (q) => + q.eq("status", "pending_checks"), + ) + .order("asc") + .take(25); - if (!attempt) return null; - if (attempt.status !== "pending_checks") { - throw new ConvexError(`Publish attempt is ${attempt.status}, not pending checks.`); - } - if (args.kind && attempt.kind !== args.kind) { - throw new ConvexError("Publish attempt kind does not match worker claim."); - } - if (args.slug && attempt.slug !== args.slug) { - throw new ConvexError("Publish attempt slug does not match worker claim."); - } - if (args.version && attempt.version !== args.version) { - throw new ConvexError("Publish attempt version does not match worker claim."); - } - if ((attempt.checkClaimExpiresAt ?? 0) > now && attempt.checkClaimId !== args.claimId) { - throw new ConvexError("Publish attempt checks are already claimed."); - } - - const checkClaimExpiresAt = now + CHECK_CLAIM_LEASE_MS; - await ctx.db.patch(attempt._id, { - checkClaimId: args.claimId, - checkClaimedAt: now, - checkClaimExpiresAt, - checkClaimLastError: undefined, - updatedAt: now, - }); - - let existingClawscanAnalysis: unknown; - if (attempt.kind === "skill" && attempt.skillVersionId) { - const version = await ctx.db.get(attempt.skillVersionId); - if (version?.fingerprint === attempt.artifactFingerprint) { - existingClawscanAnalysis = reusableClawscanAnalysis(version.llmAnalysis); + for (const attempt of candidates) { + if (attempt.status !== "pending_checks") { + if (args.attemptId && attempt.status === "failed") return null; + if (args.attemptId) { + throw new ConvexError(`Publish attempt is ${attempt.status}, not pending checks.`); + } + continue; } - } else if (attempt.kind === "package" && attempt.packageReleaseId) { - const release = await ctx.db.get(attempt.packageReleaseId); - if (release?.integritySha256 === attempt.artifactFingerprint) { - existingClawscanAnalysis = reusableClawscanAnalysis(release.llmAnalysis); + if (args.kind && attempt.kind !== args.kind) { + if (args.attemptId) { + throw new ConvexError("Publish attempt kind does not match worker claim."); + } + continue; } - } + if (args.slug && attempt.slug !== args.slug) { + if (args.attemptId) { + throw new ConvexError("Publish attempt slug does not match worker claim."); + } + continue; + } + if (args.version && attempt.version !== args.version) { + if (args.attemptId) { + throw new ConvexError("Publish attempt version does not match worker claim."); + } + continue; + } + if ((attempt.checkClaimExpiresAt ?? 0) > now && attempt.checkClaimId !== args.claimId) { + if (args.attemptId) { + throw new ConvexError("Publish attempt checks are already claimed."); + } + continue; + } + if (await terminalizeUnavailableStagedTarget(ctx, attempt, now)) continue; - return { - attemptId: attempt._id, - status: attempt.status, - claimId: args.claimId, - kind: attempt.kind, - userId: attempt.userId, - ownerUserId: attempt.ownerUserId, - ownerPublisherId: attempt.ownerPublisherId, - sourceOwnerPublisherId: attempt.sourceOwnerPublisherId, - skillId: attempt.skillId, - versionId: attempt.skillVersionId, - packageId: attempt.packageId, - releaseId: attempt.packageReleaseId, - slug: attempt.slug, - displayName: attempt.displayName, - version: attempt.version, - artifactFingerprint: attempt.artifactFingerprint, - files: attempt.files, - ...(attempt.kind === "skill" - ? { - scanContext: buildSkillAttemptScanContext(attempt), - } - : { - clawpackStorageId: publishAttemptClawpackStorageId(attempt), - scanContext: buildPackageAttemptScanContext(attempt), - }), - ...(existingClawscanAnalysis ? { existingClawscanAnalysis } : {}), - checkClaimExpiresAt, - createdAt: attempt.createdAt, - }; + const checkClaimExpiresAt = now + CHECK_CLAIM_LEASE_MS; + await ctx.db.patch(attempt._id, { + checkClaimId: args.claimId, + checkClaimedAt: now, + checkClaimExpiresAt, + checkClaimLastError: undefined, + updatedAt: now, + }); + + let existingClawscanAnalysis: unknown; + if (attempt.kind === "skill" && attempt.skillVersionId) { + const version = await ctx.db.get(attempt.skillVersionId); + if (version?.fingerprint === attempt.artifactFingerprint) { + existingClawscanAnalysis = reusableClawscanAnalysis(version.llmAnalysis); + } + } else if (attempt.kind === "package" && attempt.packageReleaseId) { + const release = await ctx.db.get(attempt.packageReleaseId); + if (release?.integritySha256 === attempt.artifactFingerprint) { + existingClawscanAnalysis = reusableClawscanAnalysis(release.llmAnalysis); + } + } + + return { + attemptId: attempt._id, + status: attempt.status, + claimId: args.claimId, + kind: attempt.kind, + userId: attempt.userId, + ownerUserId: attempt.ownerUserId, + ownerPublisherId: attempt.ownerPublisherId, + sourceOwnerPublisherId: attempt.sourceOwnerPublisherId, + skillId: attempt.skillId, + versionId: attempt.skillVersionId, + packageId: attempt.packageId, + releaseId: attempt.packageReleaseId, + slug: attempt.slug, + displayName: attempt.displayName, + version: attempt.version, + artifactFingerprint: attempt.artifactFingerprint, + files: attempt.files, + ...(attempt.kind === "skill" + ? { + scanContext: buildSkillAttemptScanContext(attempt), + } + : { + clawpackStorageId: publishAttemptClawpackStorageId(attempt), + scanContext: buildPackageAttemptScanContext(attempt), + }), + ...(existingClawscanAnalysis ? { existingClawscanAnalysis } : {}), + checkClaimExpiresAt, + createdAt: attempt.createdAt, + }; + } + return null; }, }); @@ -772,68 +831,79 @@ export const claimReadyPublishAttemptFinalizationRetryInternal = internalMutatio }, handler: async (ctx, args) => { const now = Date.now(); - const attempt = args.attemptId - ? await ctx.db.get(args.attemptId) - : ( - await ctx.db - .query("publishAttempts") - .withIndex("by_status_and_created", (q) => q.eq("status", "ready_to_finalize")) - .order("asc") - .take(25) - ).find((candidate) => { - if ((candidate.checkClaimExpiresAt ?? 0) > now) return false; - if (args.kind && candidate.kind !== args.kind) return false; - if (args.slug && candidate.slug !== args.slug) return false; - if (args.version && candidate.version !== args.version) return false; - return true; - }); + const targetedAttempt = args.attemptId ? await ctx.db.get(args.attemptId) : null; + const candidates = args.attemptId + ? targetedAttempt + ? [targetedAttempt] + : [] + : await ctx.db + .query("publishAttempts") + .withIndex("by_status_and_created", (q) => q.eq("status", "ready_to_finalize")) + .order("asc") + .take(25); - if (!attempt) return null; - if (attempt.status !== "ready_to_finalize") { - return null; - } - if (args.kind && attempt.kind !== args.kind) { - throw new ConvexError("Publish attempt kind does not match worker claim."); - } - if (args.slug && attempt.slug !== args.slug) { - throw new ConvexError("Publish attempt slug does not match worker claim."); - } - if (args.version && attempt.version !== args.version) { - throw new ConvexError("Publish attempt version does not match worker claim."); - } - if ((attempt.checkClaimExpiresAt ?? 0) > now && attempt.checkClaimId !== args.claimId) { - throw new ConvexError("Publish attempt finalization retry is already claimed."); - } + for (const attempt of candidates) { + if (attempt.status !== "ready_to_finalize") { + if (args.attemptId) return null; + continue; + } + if (args.kind && attempt.kind !== args.kind) { + if (args.attemptId) { + throw new ConvexError("Publish attempt kind does not match worker claim."); + } + continue; + } + if (args.slug && attempt.slug !== args.slug) { + if (args.attemptId) { + throw new ConvexError("Publish attempt slug does not match worker claim."); + } + continue; + } + if (args.version && attempt.version !== args.version) { + if (args.attemptId) { + throw new ConvexError("Publish attempt version does not match worker claim."); + } + continue; + } + if ((attempt.checkClaimExpiresAt ?? 0) > now && attempt.checkClaimId !== args.claimId) { + if (args.attemptId) { + throw new ConvexError("Publish attempt finalization retry is already claimed."); + } + continue; + } + if (await terminalizeUnavailableStagedTarget(ctx, attempt, now)) continue; - await ctx.db.patch(attempt._id, { - checkClaimId: args.claimId, - checkClaimedAt: now, - checkClaimExpiresAt: now + CHECK_CLAIM_LEASE_MS, - checkClaimLastError: undefined, - updatedAt: now, - }); + await ctx.db.patch(attempt._id, { + checkClaimId: args.claimId, + checkClaimedAt: now, + checkClaimExpiresAt: now + CHECK_CLAIM_LEASE_MS, + checkClaimLastError: undefined, + updatedAt: now, + }); - return { - attemptId: attempt._id, - status: attempt.status, - claimId: args.claimId, - kind: attempt.kind, - userId: attempt.userId, - ownerUserId: attempt.ownerUserId, - ownerPublisherId: attempt.ownerPublisherId, - sourceOwnerPublisherId: attempt.sourceOwnerPublisherId, - skillId: attempt.skillId, - versionId: attempt.skillVersionId, - packageId: attempt.packageId, - releaseId: attempt.packageReleaseId, - slug: attempt.slug, - displayName: attempt.displayName, - version: attempt.version, - artifactFingerprint: attempt.artifactFingerprint, - files: [], - checkClaimExpiresAt: now + CHECK_CLAIM_LEASE_MS, - createdAt: attempt.createdAt, - }; + return { + attemptId: attempt._id, + status: attempt.status, + claimId: args.claimId, + kind: attempt.kind, + userId: attempt.userId, + ownerUserId: attempt.ownerUserId, + ownerPublisherId: attempt.ownerPublisherId, + sourceOwnerPublisherId: attempt.sourceOwnerPublisherId, + skillId: attempt.skillId, + versionId: attempt.skillVersionId, + packageId: attempt.packageId, + releaseId: attempt.packageReleaseId, + slug: attempt.slug, + displayName: attempt.displayName, + version: attempt.version, + artifactFingerprint: attempt.artifactFingerprint, + files: [], + checkClaimExpiresAt: now + CHECK_CLAIM_LEASE_MS, + createdAt: attempt.createdAt, + }; + } + return null; }, }); @@ -1201,7 +1271,7 @@ export const completePrePublicationChecks: ReturnType = action({ )) as { attemptId: Id<"publishAttempts">; kind: "skill" | "package"; - status: "blocked" | "pending_checks" | "ready_to_finalize"; + status: "blocked" | "failed" | "pending_checks" | "ready_to_finalize"; }; if (completed.status !== "ready_to_finalize") return completed;