diff --git a/convex/httpApiV1.handlers.test.ts b/convex/httpApiV1.handlers.test.ts index a7c2b751..9ff26fb7 100644 --- a/convex/httpApiV1.handlers.test.ts +++ b/convex/httpApiV1.handlers.test.ts @@ -405,6 +405,7 @@ describe("httpApiV1 handlers", () => { body: JSON.stringify({ handle: "Target", slugs: ["a", "b"], + versionsBySlug: { a: "1.0.0", b: "1.1.0" }, forceOverwriteSquatter: true, }), }), @@ -415,10 +416,42 @@ describe("httpApiV1 handlers", () => { ownerHandle: "target", ownerUserId: "users:target", slugs: ["a", "b"], + versionsBySlug: { a: "1.0.0", b: "1.1.0" }, forceOverwriteSquatter: true, }); }); + it("users/restore requires a backup version for every slug", async () => { + const runAction = vi.fn(); + const runMutation = vi.fn(async (_mutation: unknown, args: Record) => { + if (isRateLimitArgs(args)) return okRate(); + return { ok: true }; + }); + const runQuery = vi.fn(); + vi.mocked(requireApiTokenUser).mockResolvedValue({ + userId: "users:admin", + user: { _id: "users:admin", role: "admin" }, + } as never); + + const response = await __handlers.usersPostRouterV1Handler( + makeCtx({ runQuery, runAction, runMutation }), + new Request("https://example.com/api/v1/users/restore", { + method: "POST", + body: JSON.stringify({ + handle: "Target", + slugs: ["a", "b"], + versionsBySlug: { a: "1.0.0" }, + forceOverwriteSquatter: true, + }), + }), + ); + + expect(response.status).toBe(400); + expect(await response.text()).toBe("Missing backup version for slug b"); + expect(runQuery).not.toHaveBeenCalled(); + expect(runAction).not.toHaveBeenCalled(); + }); + it("skills export allows authenticated non-admin users at the key rate limit", async () => { vi.mocked(requireApiTokenUser).mockResolvedValue({ userId: "users:actor", diff --git a/convex/httpApiV1/usersV1.ts b/convex/httpApiV1/usersV1.ts index ea12ea75..9ec41cb7 100644 --- a/convex/httpApiV1/usersV1.ts +++ b/convex/httpApiV1/usersV1.ts @@ -557,7 +557,7 @@ export async function usersGetRouterV1Handler(ctx: ActionCtx, request: Request) /** * POST /api/v1/users/restore * Admin-only: restore skills from registry artifact backup for a user. - * Body: { handle: string, slugs: string[], forceOverwriteSquatter?: boolean } + * Body: { handle: string, slugs: string[], versionsBySlug: Record, forceOverwriteSquatter?: boolean } */ async function handleAdminRestore( ctx: ActionCtx, @@ -575,6 +575,20 @@ async function handleAdminRestore( if (slugs.length === 0) return text("Missing slugs array", 400, headers); if (slugs.length > 100) return text("Too many slugs (max 100)", 400, headers); + const versionsBySlug = + payload.versionsBySlug && typeof payload.versionsBySlug === "object" + ? Object.fromEntries( + Object.entries(payload.versionsBySlug).filter( + (entry): entry is [string, string] => + typeof entry[0] === "string" && typeof entry[1] === "string", + ), + ) + : undefined; + if (!versionsBySlug) return text("Missing versionsBySlug", 400, headers); + const missingVersionSlug = slugs.find((slug) => !versionsBySlug[slug]?.trim()); + if (missingVersionSlug) { + return text(`Missing backup version for slug ${missingVersionSlug}`, 400, headers); + } const forceOverwriteSquatter = Boolean(payload.forceOverwriteSquatter); const targetUser = await ctx.runQuery(api.users.getByHandle, { handle }); @@ -588,6 +602,7 @@ async function handleAdminRestore( ownerHandle: handle, ownerUserId: targetUser._id, slugs, + versionsBySlug, forceOverwriteSquatter, }, ); diff --git a/convex/lib/registryArtifactBackup.test.ts b/convex/lib/registryArtifactBackup.test.ts index 9f9e6f33..e06fdcc2 100644 --- a/convex/lib/registryArtifactBackup.test.ts +++ b/convex/lib/registryArtifactBackup.test.ts @@ -6,7 +6,6 @@ import { backupSkillVersionToObjectStorage, buildPackageReleaseBackupManifest, buildSkillVersionBackupManifest, - fetchSkillBackupIndex, getRegistryArtifactBackupSettings, readRegistryArtifactBackupObject, } from "./registryArtifactBackup"; @@ -79,7 +78,6 @@ describe("registry artifact backup settings", () => { expect(manifest).toMatchObject({ skillRoot: "skills/openclaw-team/demo-skill", versionRoot: "skills/openclaw-team/demo-skill/1%2E2%2E3", - indexPath: "skills/openclaw-team/demo-skill/_index.json", metaPath: "skills/openclaw-team/demo-skill/1%2E2%2E3/_meta.json", fileObjects: [ { @@ -153,7 +151,6 @@ describe("registry artifact backup settings", () => { artifactPath: "packages/openclaw-team/%40openclaw%2Fdemo-plugin/1%2E2%2E3/demo-plugin-1.2.3.tgz", metaPath: "packages/openclaw-team/%40openclaw%2Fdemo-plugin/1%2E2%2E3/_meta.json", - indexPath: "packages/openclaw-team/%40openclaw%2Fdemo-plugin/_index.json", meta: { kind: "packageRelease", restore: { @@ -195,181 +192,6 @@ describe("registry artifact backup settings", () => { ).toThrow("Invalid package backup artifact filename"); }); - it("keeps skill index latest pointers on the greatest semver version without an explicit latest", () => { - const backport = buildSkillVersionBackupManifest({ - root: "skills", - ownerHandle: "OpenClaw Team", - versionId: "skillVersions:demo-1" as Id<"skillVersions">, - slug: "demo-skill", - displayName: "Demo Skill", - version: "1.0.0", - publishedAt: 1_900_000_000_000, - files: [], - }); - - const index = __registryArtifactBackupTestInternals.buildSkillIndexFile(backport, { - kind: "skill", - owner: "openclaw-team", - slug: "demo-skill", - displayName: "Demo Skill", - latest: { - version: "2.0.0", - publishedAt: 1_800_000_000_000, - versionId: "skillVersions:demo-2" as Id<"skillVersions">, - path: "skills/openclaw-team/demo-skill/2%2E0%2E0/_meta.json", - }, - versions: [], - }); - - expect(index.latest.version).toBe("2.0.0"); - expect(index.versions.map((version) => version.version)).toEqual(["2.0.0", "1.0.0"]); - }); - - it("preserves explicit skill latest pointers after a rollback", () => { - const rolledBackLatest = buildSkillVersionBackupManifest({ - root: "skills", - ownerHandle: "OpenClaw Team", - versionId: "skillVersions:demo-1" as Id<"skillVersions">, - slug: "demo-skill", - displayName: "Demo Skill", - version: "1.0.0", - isLatest: true, - publishedAt: 1_700_000_000_000, - files: [], - }); - - const index = __registryArtifactBackupTestInternals.buildSkillIndexFile(rolledBackLatest, { - kind: "skill", - owner: "openclaw-team", - slug: "demo-skill", - displayName: "Demo Skill", - latest: { - version: "2.0.0", - isLatest: true, - publishedAt: 1_800_000_000_000, - versionId: "skillVersions:demo-2" as Id<"skillVersions">, - path: "skills/openclaw-team/demo-skill/2%2E0%2E0/_meta.json", - }, - versions: [], - }); - - expect(index.latest).toMatchObject({ - version: "1.0.0", - isLatest: true, - versionId: "skillVersions:demo-1", - }); - expect(index.versions.find((version) => version.version === "2.0.0")?.isLatest).toBe(false); - }); - - it("keeps package index latest pointers on explicit latest release markers", () => { - const backport = buildPackageReleaseBackupManifest({ - root: "packages", - ownerHandle: "OpenClaw Team", - packageId: "packages:demo" as Id<"packages">, - releaseId: "packageReleases:demo-1" as Id<"packageReleases">, - packageName: "@openclaw/demo-plugin", - normalizedName: "@openclaw/demo-plugin", - displayName: "Demo Plugin", - family: "code-plugin", - version: "1.0.0", - isLatest: false, - publishedAt: 1_900_000_000_000, - artifactKind: "npm-pack", - artifactSha256: "sha256:artifact", - artifactSize: 42, - artifactFormat: "tgz", - files: [], - }); - - const index = __registryArtifactBackupTestInternals.buildPackageIndexFile(backport, { - kind: "package", - owner: "openclaw-team", - packageName: "@openclaw/demo-plugin", - normalizedName: "@openclaw/demo-plugin", - displayName: "Demo Plugin", - family: "code-plugin", - latest: { - version: "2.0.0", - isLatest: true, - publishedAt: 1_800_000_000_000, - packageId: "packages:demo" as Id<"packages">, - releaseId: "packageReleases:demo-2" as Id<"packageReleases">, - path: "packages/openclaw-team/%40openclaw%2Fdemo-plugin/2%2E0%2E0/_meta.json", - }, - versions: [], - }); - - expect(index.latest).toMatchObject({ - version: "2.0.0", - isLatest: true, - releaseId: "packageReleases:demo-2", - }); - }); - - it("keeps full version catalogs in skill and package indexes", () => { - const skillBackup = buildSkillVersionBackupManifest({ - root: "skills", - ownerHandle: "OpenClaw Team", - versionId: "skillVersions:demo-new" as Id<"skillVersions">, - slug: "demo-skill", - displayName: "Demo Skill", - version: "1001.0.0", - publishedAt: 1_800_000_001_000, - files: [], - }); - const packageBackup = buildPackageReleaseBackupManifest({ - root: "packages", - ownerHandle: "OpenClaw Team", - packageId: "packages:demo" as Id<"packages">, - releaseId: "packageReleases:demo-new" as Id<"packageReleases">, - packageName: "@openclaw/demo-plugin", - normalizedName: "@openclaw/demo-plugin", - displayName: "Demo Plugin", - family: "code-plugin", - version: "1001.0.0", - publishedAt: 1_800_000_001_000, - files: [], - }); - const existingSkillVersions = Array.from({ length: 1001 }, (_, index) => ({ - version: `${index}.0.0`, - publishedAt: 1_800_000_000_000 - index, - versionId: `skillVersions:demo-${index}` as Id<"skillVersions">, - path: `skills/openclaw-team/demo-skill/${index}%2E0%2E0/_meta.json`, - })); - const existingPackageVersions = Array.from({ length: 1001 }, (_, index) => ({ - version: `${index}.0.0`, - publishedAt: 1_800_000_000_000 - index, - packageId: "packages:demo" as Id<"packages">, - releaseId: `packageReleases:demo-${index}` as Id<"packageReleases">, - path: `packages/openclaw-team/%40openclaw%2Fdemo-plugin/${index}%2E0%2E0/_meta.json`, - })); - - const skillIndex = __registryArtifactBackupTestInternals.buildSkillIndexFile(skillBackup, { - kind: "skill", - owner: "openclaw-team", - slug: "demo-skill", - displayName: "Demo Skill", - latest: existingSkillVersions[0]!, - versions: existingSkillVersions, - }); - const packageIndex = __registryArtifactBackupTestInternals.buildPackageIndexFile( - packageBackup, - { - kind: "package", - owner: "openclaw-team", - packageName: "@openclaw/demo-plugin", - normalizedName: "@openclaw/demo-plugin", - displayName: "Demo Plugin", - family: "code-plugin", - latest: existingPackageVersions[0]!, - versions: existingPackageVersions, - }, - ); - - expect(skillIndex.versions).toHaveLength(1002); - expect(packageIndex.versions).toHaveLength(1002); - }); - it("uses lossless path encoding to avoid package and version collisions", () => { expect(__registryArtifactBackupTestInternals.encodeBackupPathSegment("@openclaw/demo")).toBe( "%40openclaw%2Fdemo", @@ -421,28 +243,11 @@ describe("registry artifact backup settings", () => { ]); }); - it("reads skill indexes and object bytes from object storage", async () => { + it("reads object bytes from object storage", async () => { vi.stubGlobal( "fetch", vi.fn(async (url: URL | string, init?: RequestInit) => { const key = objectKey(String(url)); - if (init?.method === "GET" && key === "skills/openclaw-team/demo-skill/_index.json") { - return response( - 200, - JSON.stringify({ - kind: "skill", - owner: "openclaw-team", - slug: "demo-skill", - displayName: "Demo Skill", - latest: { - version: "1.2.3", - publishedAt: 1_700_000_000_000, - path: "skills/openclaw-team/demo-skill/1%2E2%2E3/_meta.json", - }, - versions: [], - }), - ); - } if ( init?.method === "GET" && key === "skills/openclaw-team/demo-skill/1%2E2%2E3/SKILL.md" @@ -453,17 +258,15 @@ describe("registry artifact backup settings", () => { }), ); - const index = await fetchSkillBackupIndex(makeContext(), "OpenClaw Team", "demo-skill"); const bytes = await readRegistryArtifactBackupObject( makeContext(), "skills/openclaw-team/demo-skill/1%2E2%2E3/SKILL.md", ); - expect(index?.latest.version).toBe("1.2.3"); expect(Buffer.from(bytes!).toString("utf8")).toBe("hello skill"); }); - it("writes skill files, version metadata, and the skill index to object storage", async () => { + it("writes skill files and version metadata to object storage", async () => { const calls: Array<{ method: string; url: string; body: string }> = []; vi.stubGlobal( "fetch", @@ -471,7 +274,6 @@ describe("registry artifact backup settings", () => { const method = init?.method ?? "GET"; const body = await requestBodyText(init?.body); calls.push({ method, url: String(url), body }); - if (method === "GET") return response(404, ""); return response(200, ""); }), ); @@ -502,141 +304,15 @@ describe("registry artifact backup settings", () => { expect(calls.map((call) => [call.method, objectKey(call.url)])).toEqual([ ["PUT", "skills/openclaw-team/demo-skill/1%2E2%2E3/SKILL.md"], ["PUT", "skills/openclaw-team/demo-skill/1%2E2%2E3/_meta.json"], - ["GET", "skills/openclaw-team/demo-skill/_index.json"], - ["PUT", "skills/openclaw-team/demo-skill/_index.json"], ]); expect(JSON.parse(calls[1].body)).toMatchObject({ kind: "skillVersion", version: "1.2.3", metadata: { files: [{ path: "SKILL.md", sha256: "sha256:skill" }] }, }); - expect(JSON.parse(calls[3].body)).toMatchObject({ - kind: "skill", - latest: { version: "1.2.3" }, - versions: [{ version: "1.2.3" }], - }); }); - it("retries skill index writes when another backup updates the index first", async () => { - const calls: Array<{ method: string; url: string; body: string; ifMatch?: string }> = []; - vi.stubGlobal( - "fetch", - vi.fn(async (url: URL | string, init?: RequestInit) => { - const method = init?.method ?? "GET"; - const key = objectKey(String(url)); - const body = await requestBodyText(init?.body); - calls.push({ - method, - url: String(url), - body, - ifMatch: headerValue(init?.headers, "if-match"), - }); - - if (method === "GET" && key === "skills/openclaw-team/demo-skill/_index.json") { - const indexGetCount = calls.filter( - (call) => - call.method === "GET" && - objectKey(call.url) === "skills/openclaw-team/demo-skill/_index.json", - ).length; - if (indexGetCount === 1) { - return response( - 200, - JSON.stringify({ - kind: "skill", - owner: "openclaw-team", - slug: "demo-skill", - displayName: "Demo Skill", - latest: { - version: "1.0.0", - publishedAt: 1_600_000_000_000, - path: "skills/openclaw-team/demo-skill/1%2E0%2E0/_meta.json", - }, - versions: [ - { - version: "1.0.0", - publishedAt: 1_600_000_000_000, - path: "skills/openclaw-team/demo-skill/1%2E0%2E0/_meta.json", - }, - ], - }), - { etag: '"old-index"' }, - ); - } - return response( - 200, - JSON.stringify({ - kind: "skill", - owner: "openclaw-team", - slug: "demo-skill", - displayName: "Demo Skill", - latest: { - version: "2.0.0", - publishedAt: 1_800_000_000_000, - path: "skills/openclaw-team/demo-skill/2%2E0%2E0/_meta.json", - }, - versions: [ - { - version: "2.0.0", - publishedAt: 1_800_000_000_000, - path: "skills/openclaw-team/demo-skill/2%2E0%2E0/_meta.json", - }, - { - version: "1.0.0", - publishedAt: 1_600_000_000_000, - path: "skills/openclaw-team/demo-skill/1%2E0%2E0/_meta.json", - }, - ], - }), - { etag: '"new-index"' }, - ); - } - - if (method === "PUT" && key === "skills/openclaw-team/demo-skill/_index.json") { - return headerValue(init?.headers, "if-match") === '"old-index"' - ? response(412, "precondition failed") - : response(200, ""); - } - - return response(200, ""); - }), - ); - - await backupSkillVersionToObjectStorage( - makeStorageCtx({ "storage:skill": "hello skill" }) as never, - { - root: "skills", - ownerHandle: "OpenClaw Team", - versionId: "skillVersions:demo-1.2" as Id<"skillVersions">, - slug: "demo-skill", - displayName: "Demo Skill", - version: "1.2.3", - publishedAt: 1_700_000_000_000, - files: [ - { - path: "SKILL.md", - size: 11, - storageId: "storage:skill" as Id<"_storage">, - sha256: "sha256:skill", - contentType: "text/markdown", - }, - ], - }, - makeContext(), - ); - - const indexPuts = calls.filter( - (call) => - call.method === "PUT" && - objectKey(call.url) === "skills/openclaw-team/demo-skill/_index.json", - ); - expect(indexPuts.map((call) => call.ifMatch)).toEqual(['"old-index"', '"new-index"']); - expect(JSON.parse(indexPuts[1].body)).toMatchObject({ - latest: { version: "2.0.0" }, - versions: [{ version: "2.0.0" }, { version: "1.2.3" }, { version: "1.0.0" }], - }); - }); - - it("writes package artifacts, version metadata, and the package index to object storage", async () => { + it("writes package artifacts and version metadata to object storage", async () => { const calls: Array<{ method: string; url: string; body: string }> = []; vi.stubGlobal( "fetch", @@ -644,7 +320,6 @@ describe("registry artifact backup settings", () => { const method = init?.method ?? "GET"; const body = await requestBodyText(init?.body); calls.push({ method, url: String(url), body }); - if (method === "GET") return response(404, ""); return response(200, ""); }), ); @@ -674,17 +349,11 @@ describe("registry artifact backup settings", () => { expect(calls.map((call) => [call.method, objectKey(call.url)])).toEqual([ ["PUT", "packages/openclaw-team/%40openclaw%2Fdemo-plugin/1%2E2%2E3/demo-plugin-1.2.3.tgz"], ["PUT", "packages/openclaw-team/%40openclaw%2Fdemo-plugin/1%2E2%2E3/_meta.json"], - ["GET", "packages/openclaw-team/%40openclaw%2Fdemo-plugin/_index.json"], - ["PUT", "packages/openclaw-team/%40openclaw%2Fdemo-plugin/_index.json"], ]); expect(JSON.parse(calls[1].body)).toMatchObject({ kind: "packageRelease", artifact: { path: "demo-plugin-1.2.3.tgz", sha256: "sha256:artifact" }, }); - expect(JSON.parse(calls[3].body)).toMatchObject({ - kind: "package", - latest: { version: "1.2.3", releaseId: "packageReleases:demo-1" }, - }); }); }); @@ -743,7 +412,3 @@ async function requestBodyText(body: BodyInit | null | undefined) { if (!body) return ""; return Buffer.from(await new Response(body).arrayBuffer()).toString("utf8"); } - -function headerValue(headers: HeadersInit | undefined, name: string) { - return new Headers(headers).get(name) ?? undefined; -} diff --git a/convex/lib/registryArtifactBackup.ts b/convex/lib/registryArtifactBackup.ts index fd3318d8..6d3c4e4b 100644 --- a/convex/lib/registryArtifactBackup.ts +++ b/convex/lib/registryArtifactBackup.ts @@ -1,7 +1,6 @@ "use node"; import { createHash, createHmac } from "node:crypto"; -import semver from "semver"; import type { Id } from "../_generated/dataModel"; import type { ActionCtx } from "../_generated/server"; import { validateFilePath } from "./skillZip"; @@ -9,10 +8,6 @@ import { validateFilePath } from "./skillZip"; const DEFAULT_SKILLS_ROOT = "skills"; const DEFAULT_PACKAGES_ROOT = "packages"; const META_FILENAME = "_meta.json"; -const INDEX_FILENAME = "_index.json"; -const MAX_INDEX_WRITE_ATTEMPTS = 5; -const MIN_INDEX_WRITE_RETRY_DELAY_MS = 25; -const MAX_INDEX_WRITE_RETRY_DELAY_MS = 250; type BackupFile = { path: string; @@ -64,47 +59,6 @@ type PackageBackupParams = { files: Array<{ path: string; size: number; sha256: string }>; }; -type IndexWriteOptions = { - withIndexWrite?: (indexPath: string, write: () => Promise) => Promise; -}; - -type VersionIndexEntry = { - version: string; - isLatest?: boolean; - publishedAt: number; - path: string; -}; - -type SkillIndexEntry = VersionIndexEntry & { - skillId?: Id<"skills">; - versionId?: Id<"skillVersions">; -}; - -type PackageIndexEntry = VersionIndexEntry & { - packageId: Id<"packages">; - releaseId: Id<"packageReleases">; -}; - -type SkillIndexFile = { - kind: "skill"; - owner: string; - slug: string; - displayName: string; - latest: SkillIndexEntry; - versions: SkillIndexEntry[]; -}; - -type PackageIndexFile = { - kind: "package"; - owner: string; - packageName: string; - normalizedName: string; - displayName: string; - family: PackageBackupParams["family"]; - latest: PackageIndexEntry; - versions: PackageIndexEntry[]; -}; - export type RegistryArtifactBackupContext = RegistryArtifactBackupSettings; export type RegistryArtifactBackupSettings = { @@ -155,7 +109,6 @@ export async function backupSkillVersionToObjectStorage( ctx: Pick, params: SkillBackupParams & { root?: string }, context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), - options: IndexWriteOptions = {}, ) { const planned = buildSkillVersionBackupManifest({ root: params.root ?? context.skillsRoot, @@ -170,19 +123,12 @@ export async function backupSkillVersionToObjectStorage( } await putJsonObject(context, planned.metaPath, planned.meta); - await writeMergedJsonIndex( - context, - planned.indexPath, - (existingIndex: SkillIndexFile | null) => buildSkillIndexFile(planned, existingIndex), - options, - ); } export async function backupPackageReleaseToObjectStorage( ctx: Pick, params: PackageBackupParams & { artifactStorageId: Id<"_storage">; root?: string }, context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), - options: IndexWriteOptions = {}, ) { const planned = buildPackageReleaseBackupManifest({ root: params.root ?? context.packagesRoot, @@ -194,86 +140,6 @@ export async function backupPackageReleaseToObjectStorage( }); await putJsonObject(context, planned.metaPath, planned.meta); - await writeMergedJsonIndex( - context, - planned.indexPath, - (existingIndex: PackageIndexFile | null) => buildPackageIndexFile(planned, existingIndex), - options, - ); -} - -export async function repairSkillVersionBackupIndex( - _ctx: Pick, - params: SkillBackupParams & { root?: string }, - context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), - options: IndexWriteOptions = {}, -) { - await repairSkillVersionBackupIndexes(_ctx, [params], context, options); -} - -export async function repairSkillVersionBackupIndexes( - _ctx: Pick, - params: Array, - context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), - options: IndexWriteOptions = {}, -) { - if (params.length === 0) return; - const planned = params.map((item) => - buildSkillVersionBackupManifest({ - root: item.root ?? context.skillsRoot, - ...item, - }), - ); - const [first, ...rest] = planned; - if (!first) return; - const indexPath = sharedIndexPath(planned.map((item) => item.indexPath)); - await writeMergedJsonIndex( - context, - indexPath, - (existingIndex: SkillIndexFile | null) => - rest.reduce( - (nextIndex, plannedItem) => buildSkillIndexFile(plannedItem, nextIndex), - buildSkillIndexFile(first, existingIndex), - ), - options, - ); -} - -export async function repairPackageReleaseBackupIndex( - _ctx: Pick, - params: PackageBackupParams & { root?: string }, - context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), - options: IndexWriteOptions = {}, -) { - await repairPackageReleaseBackupIndexes(_ctx, [params], context, options); -} - -export async function repairPackageReleaseBackupIndexes( - _ctx: Pick, - params: Array, - context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), - options: IndexWriteOptions = {}, -) { - if (params.length === 0) return; - const planned = params.map((item) => - buildPackageReleaseBackupManifest({ - root: item.root ?? context.packagesRoot, - ...item, - }), - ); - const [first, ...rest] = planned; - if (!first) return; - const indexPath = sharedIndexPath(planned.map((item) => item.indexPath)); - await writeMergedJsonIndex( - context, - indexPath, - (existingIndex: PackageIndexFile | null) => - rest.reduce( - (nextIndex, plannedItem) => buildPackageIndexFile(plannedItem, nextIndex), - buildPackageIndexFile(first, existingIndex), - ), - options, - ); } export async function fetchSkillVersionBackupMeta( @@ -289,41 +155,6 @@ export async function fetchSkillVersionBackupMeta( return getJsonObject["meta"]>(context, path); } -async function writeMergedJsonIndex( - context: RegistryArtifactBackupContext, - indexPath: string, - buildNext: (existing: T | null) => T, - options: IndexWriteOptions, -) { - const write = () => putMergedJsonIndex(context, indexPath, buildNext); - if (options.withIndexWrite) { - return options.withIndexWrite(indexPath, write); - } - return write(); -} - -export async function fetchSkillBackupIndex( - context: RegistryArtifactBackupContext, - ownerHandle: string, - slug: string, -) { - const owner = normalizeOwner(ownerHandle); - const path = `${context.skillsRoot}/${owner}/${slug}/${INDEX_FILENAME}`; - return getJsonObject(context, path); -} - -export async function fetchPackageBackupIndex( - context: RegistryArtifactBackupContext, - ownerHandle: string, - normalizedName: string, -) { - const owner = normalizeOwner(ownerHandle); - const path = `${context.packagesRoot}/${owner}/${encodeBackupPathSegment( - normalizedName, - )}/${INDEX_FILENAME}`; - return getJsonObject(context, path); -} - export async function fetchPackageReleaseBackupMeta( context: RegistryArtifactBackupContext, ownerHandle: string, @@ -356,7 +187,6 @@ export function buildSkillVersionBackupManifest(params: SkillBackupParams & { ro const skillRoot = `${params.root}/${owner}/${params.slug}`; const versionRoot = `${skillRoot}/${versionSegment}`; const metaPath = `${versionRoot}/${META_FILENAME}`; - const indexPath = `${skillRoot}/${INDEX_FILENAME}`; const files = params.files.map((file) => { if (!validateFilePath(file.path)) { throw new Error(`Invalid skill backup file path: ${file.path}`); @@ -393,7 +223,6 @@ export function buildSkillVersionBackupManifest(params: SkillBackupParams & { ro skillRoot, versionRoot, metaPath, - indexPath, fileObjects, meta, }; @@ -449,113 +278,11 @@ export function buildPackageReleaseBackupManifest(params: PackageBackupParams & releaseRoot, artifactPath: `${releaseRoot}/${artifactFileName}`, metaPath: `${releaseRoot}/${META_FILENAME}`, - indexPath: `${packageRoot}/${INDEX_FILENAME}`, meta, }; } -function buildSkillIndexFile( - planned: ReturnType, - existing: SkillIndexFile | null, -): SkillIndexFile { - const nextVersion: SkillIndexEntry = { - version: planned.meta.version, - isLatest: planned.meta.isLatest, - publishedAt: planned.meta.publishedAt, - skillId: planned.meta.restore.skillId, - versionId: planned.meta.restore.versionId, - path: planned.metaPath, - }; - const byVersion = new Map(); - for (const entry of [nextVersion, existing?.latest, ...(existing?.versions ?? [])]) { - if (entry && !byVersion.has(entry.version)) byVersion.set(entry.version, entry); - } - const mergedVersions = Array.from(byVersion.values()); - const explicitLatest = nextVersion.isLatest - ? nextVersion - : mergedVersions.find((entry) => entry.isLatest); - const versions = mergedVersions - .map((entry) => ({ - ...entry, - isLatest: explicitLatest ? entry.version === explicitLatest.version : entry.isLatest, - })) - .sort(compareSkillIndexEntriesForLatest); - const latest = explicitLatest - ? (versions.find((entry) => entry.version === explicitLatest.version) ?? explicitLatest) - : (versions[0] ?? nextVersion); - - return { - kind: "skill", - owner: planned.meta.owner, - slug: planned.meta.slug, - displayName: planned.meta.displayName, - latest, - versions, - }; -} - -function sharedIndexPath(paths: string[]) { - const [first, ...rest] = paths; - if (!first || rest.some((path) => path !== first)) { - throw new Error("Registry artifact backup bulk index repair received mixed roots"); - } - return first; -} - -function compareSkillIndexEntriesForLatest(left: SkillIndexEntry, right: SkillIndexEntry) { - const leftValid = semver.valid(left.version); - const rightValid = semver.valid(right.version); - if (leftValid && rightValid) return semver.rcompare(leftValid, rightValid); - if (leftValid) return -1; - if (rightValid) return 1; - return right.publishedAt - left.publishedAt; -} - -function buildPackageIndexFile( - planned: ReturnType, - existing: PackageIndexFile | null, -): PackageIndexFile { - const nextVersion: PackageIndexEntry = { - version: planned.meta.version, - isLatest: planned.meta.isLatest, - publishedAt: planned.meta.publishedAt, - packageId: planned.meta.restore.packageId, - releaseId: planned.meta.restore.releaseId, - path: planned.metaPath, - }; - const byRelease = new Map(); - for (const entry of [nextVersion, existing?.latest, ...(existing?.versions ?? [])]) { - if (entry && !byRelease.has(entry.releaseId)) byRelease.set(entry.releaseId, entry); - } - const mergedVersions = Array.from(byRelease.values()); - const explicitLatest = nextVersion.isLatest - ? nextVersion - : mergedVersions.find((entry) => entry.isLatest); - const versions = mergedVersions - .map((entry) => ({ - ...entry, - isLatest: explicitLatest ? entry.releaseId === explicitLatest.releaseId : entry.isLatest, - })) - .sort((a, b) => b.publishedAt - a.publishedAt); - const latest = explicitLatest - ? (versions.find((entry) => entry.releaseId === explicitLatest.releaseId) ?? explicitLatest) - : (versions[0] ?? nextVersion); - - return { - kind: "package", - owner: planned.meta.owner, - packageName: planned.meta.packageName, - normalizedName: planned.meta.normalizedName, - displayName: planned.meta.displayName, - family: planned.meta.family, - latest, - versions, - }; -} - export const __registryArtifactBackupTestInternals = { - buildPackageIndexFile, - buildSkillIndexFile, encodeBackupPathSegment, }; @@ -614,63 +341,13 @@ async function putJsonObject(context: RegistryArtifactBackupContext, key: string } async function getJsonObject(context: RegistryArtifactBackupContext, key: string) { - const result = await getJsonObjectForUpdate(context, key); - return result.value as T | null; -} - -async function getJsonObjectForUpdate(context: RegistryArtifactBackupContext, key: string) { const response = await signedFetch(context, "GET", key); - if (response.status === 404) return { value: null, etag: null }; + if (response.status === 404) return null; if (!response.ok) { const body = await response.text(); throw new Error(`Registry artifact backup GET ${key} failed: ${body}`); } - return { - value: (await response.json()) as unknown, - etag: response.headers.get("etag"), - }; -} - -async function putMergedJsonIndex( - context: RegistryArtifactBackupContext, - key: string, - buildNext: (existing: T | null) => T, -) { - for (let attempt = 1; attempt <= MAX_INDEX_WRITE_ATTEMPTS; attempt++) { - const existing = await getJsonObjectForUpdate(context, key); - const existingValue = existing.value as T | null; - if (existingValue && !existing.etag) { - throw new Error(`Registry artifact backup GET ${key} missing ETag`); - } - - const result = await putObject( - context, - key, - `${JSON.stringify(buildNext(existingValue), null, 2)}\n`, - { - contentType: "application/json; charset=utf-8", - ifMatch: existing.etag ?? undefined, - ifNoneMatch: existingValue ? undefined : "*", - allowPreconditionFailed: true, - }, - ); - if (result === "ok") return; - await sleep(indexWriteRetryDelayMs(attempt)); - } - - throw new Error(`Registry artifact backup index ${key} changed too frequently`); -} - -function indexWriteRetryDelayMs(attempt: number) { - const base = Math.min( - MAX_INDEX_WRITE_RETRY_DELAY_MS, - MIN_INDEX_WRITE_RETRY_DELAY_MS * 2 ** attempt, - ); - return base + Math.floor(Math.random() * MIN_INDEX_WRITE_RETRY_DELAY_MS); -} - -function sleep(ms: number) { - return new Promise((resolve) => setTimeout(resolve, ms)); + return (await response.json()) as T; } async function putObject( @@ -679,20 +356,13 @@ async function putObject( body: string | Uint8Array, options: { contentType?: string; - ifMatch?: string; - ifNoneMatch?: string; - allowPreconditionFailed?: boolean; } = {}, ) { const response = await signedFetch(context, "PUT", key, body, options); - if (options.allowPreconditionFailed && response.status === 412) { - return "preconditionFailed" as const; - } if (!response.ok) { const responseBody = await response.text(); throw new Error(`Registry artifact backup PUT ${key} failed: ${responseBody}`); } - return "ok" as const; } async function signedFetch( @@ -700,7 +370,7 @@ async function signedFetch( method: "GET" | "PUT", key: string, body?: string | Uint8Array, - options: { contentType?: string; ifMatch?: string; ifNoneMatch?: string } = {}, + options: { contentType?: string } = {}, ) { const now = new Date(); const bodyBytes = body === undefined ? new Uint8Array() : toBytes(body); @@ -711,8 +381,6 @@ async function signedFetch( headers.set("x-amz-content-sha256", payloadHash); headers.set("x-amz-date", amzDate(now)); if (options.contentType) headers.set("content-type", options.contentType); - if (options.ifMatch) headers.set("if-match", options.ifMatch); - if (options.ifNoneMatch) headers.set("if-none-match", options.ifNoneMatch); headers.set( "authorization", authorizationHeader(context, method, url, headers, payloadHash, now), diff --git a/convex/registryArtifactBackups.test.ts b/convex/registryArtifactBackups.test.ts index 53605423..06c04ad5 100644 --- a/convex/registryArtifactBackups.test.ts +++ b/convex/registryArtifactBackups.test.ts @@ -7,8 +7,6 @@ import { getRegistryArtifactBackupPageInternal, getPackageRegistryArtifactBackupPageInternal, releaseRegistryArtifactBackupRetryLeaseHandler, - releaseRegistryArtifactBackupIndexLeaseHandler, - tryAcquireRegistryArtifactBackupIndexLeaseHandler, tryAcquireRegistryArtifactBackupRetryLeaseHandler, } from "./registryArtifactBackups"; import { @@ -19,35 +17,13 @@ import { } from "./registryArtifactBackupsNode"; const registryBackupMocks = vi.hoisted(() => { - const normalizeOwner = (value: string) => - value - .trim() - .toLowerCase() - .replace(/^@+/, "") - .replace(/[^a-z0-9._-]/g, "-") - .replace(/-+/g, "-") - .replace(/^[._-]+|[._-]+$/g, "") || "unknown"; - const encodeBackupPathSegment = (value: string) => - encodeURIComponent(value.trim()).replace(/\./g, "%2E"); return { backupPackageReleaseToObjectStorage: vi.fn(), backupSkillVersionToObjectStorage: vi.fn(), - buildPackageReleaseBackupManifest: vi.fn((params) => ({ - indexPath: `${params.root}/${normalizeOwner(params.ownerHandle)}/${encodeBackupPathSegment(params.normalizedName || params.packageName)}/_index.json`, - })), - buildSkillVersionBackupManifest: vi.fn((params) => ({ - indexPath: `${params.root}/${normalizeOwner(params.ownerHandle)}/${params.slug}/_index.json`, - })), - fetchPackageBackupIndex: vi.fn(), fetchPackageReleaseBackupMeta: vi.fn(), - fetchSkillBackupIndex: vi.fn(), fetchSkillVersionBackupMeta: vi.fn(), getRegistryArtifactBackupContext: vi.fn(), isRegistryArtifactBackupConfigured: vi.fn(), - repairPackageReleaseBackupIndex: vi.fn(), - repairPackageReleaseBackupIndexes: vi.fn(), - repairSkillVersionBackupIndex: vi.fn(), - repairSkillVersionBackupIndexes: vi.fn(), }; }); @@ -81,92 +57,13 @@ beforeEach(() => { }; registryBackupMocks.getRegistryArtifactBackupContext.mockReturnValue(backupContext); registryBackupMocks.isRegistryArtifactBackupConfigured.mockReturnValue(true); - registryBackupMocks.backupSkillVersionToObjectStorage.mockImplementation( - async ( - _ctx: unknown, - params: { root?: string; ownerHandle: string; slug: string }, - context: typeof backupContext, - options?: { - withIndexWrite?: (indexPath: string, write: () => Promise) => Promise; - }, - ) => { - const manifest = registryBackupMocks.buildSkillVersionBackupManifest({ - root: params.root ?? context.skillsRoot, - ...params, - }); - await options?.withIndexWrite?.(manifest.indexPath, async () => undefined); - }, - ); - registryBackupMocks.backupPackageReleaseToObjectStorage.mockImplementation( - async ( - _ctx: unknown, - params: { - root?: string; - ownerHandle: string; - normalizedName: string; - packageName: string; - }, - context: typeof backupContext, - options?: { - withIndexWrite?: (indexPath: string, write: () => Promise) => Promise; - }, - ) => { - const manifest = registryBackupMocks.buildPackageReleaseBackupManifest({ - root: params.root ?? context.packagesRoot, - ...params, - }); - await options?.withIndexWrite?.(manifest.indexPath, async () => undefined); - }, - ); - registryBackupMocks.repairSkillVersionBackupIndexes.mockImplementation( - async ( - _ctx: unknown, - params: Array<{ root?: string; ownerHandle: string; slug: string }>, - context: typeof backupContext, - options?: { - withIndexWrite?: (indexPath: string, write: () => Promise) => Promise; - }, - ) => { - const first = params[0]; - if (!first) return; - const manifest = registryBackupMocks.buildSkillVersionBackupManifest({ - root: first.root ?? context.skillsRoot, - ...first, - }); - await options?.withIndexWrite?.(manifest.indexPath, async () => undefined); - }, - ); - registryBackupMocks.repairPackageReleaseBackupIndexes.mockImplementation( - async ( - _ctx: unknown, - params: Array<{ - root?: string; - ownerHandle: string; - normalizedName: string; - packageName: string; - }>, - context: typeof backupContext, - options?: { - withIndexWrite?: (indexPath: string, write: () => Promise) => Promise; - }, - ) => { - const first = params[0]; - if (!first) return; - const manifest = registryBackupMocks.buildPackageReleaseBackupManifest({ - root: first.root ?? context.packagesRoot, - ...first, - }); - await options?.withIndexWrite?.(manifest.indexPath, async () => undefined); - }, - ); + registryBackupMocks.backupSkillVersionToObjectStorage.mockResolvedValue(undefined); + registryBackupMocks.backupPackageReleaseToObjectStorage.mockResolvedValue(undefined); }); function retryLeaseRunMutation() { return vi.fn(async (_ref, args) => { if (args && typeof args === "object" && "token" in args) { - if ("indexPath" in args) { - return { acquired: true, released: true }; - } return { acquired: true, released: true }; } return undefined; @@ -225,8 +122,6 @@ describe("publish-time registry artifact backups", () => { ownerHandle: "alice", isLatest: false, }), - expect.anything(), - expect.anything(), ); }); @@ -290,8 +185,6 @@ describe("publish-time registry artifact backups", () => { displayName: "Current Package", isLatest: false, }), - expect.anything(), - expect.anything(), ); }); }); @@ -769,7 +662,7 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => { expect(runQuery.mock.calls[0]?.[1]).toMatchObject({ ignoreNextRunAt: true }); }); - it("serializes retry artifact backups with a per-index lease", async () => { + it("backs up retry artifacts without acquiring per-index leases", async () => { const jobs = [makeSkillBackupJob("demo", "skillVersions:demo")]; const skill = { ...makeSkill("skills:demo", "demo-skill"), @@ -799,18 +692,7 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => { expect(result.stats.retryJobsSucceeded).toBe(1); expect(registryBackupMocks.backupSkillVersionToObjectStorage).toHaveBeenCalledOnce(); - const indexLeaseCalls = runMutation.mock.calls.filter( - (call) => call[1]?.indexPath === "skills/alice/demo-skill/_index.json", - ); - expect(indexLeaseCalls.map((call) => call[1])).toEqual([ - expect.objectContaining({ - indexPath: "skills/alice/demo-skill/_index.json", - ttlMs: 5 * 60 * 1000, - }), - expect.objectContaining({ - indexPath: "skills/alice/demo-skill/_index.json", - }), - ]); + expect(runMutation.mock.calls.some((call) => "indexPath" in (call[1] ?? {}))).toBe(false); }); it("drains retry jobs without scanning the historical registry", async () => { @@ -960,7 +842,7 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => { ); }); - it("repairs the index without reuploading skill files when retry metadata already exists", async () => { + it("marks skill retries succeeded without reuploading when version metadata already exists", async () => { const dueJob = { _id: "registryArtifactBackupJobs:demo", targetKind: "skillVersion", @@ -1018,21 +900,13 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => { expect(result.stats.retryJobsSucceeded).toBe(1); expect(registryBackupMocks.backupSkillVersionToObjectStorage).not.toHaveBeenCalled(); - expect(registryBackupMocks.repairSkillVersionBackupIndexes).toHaveBeenCalledWith( - expect.anything(), - [ - expect.objectContaining({ - slug: "demo-skill", - version: "1.0.0", - ownerHandle: "alice", - }), - ], - expect.anything(), + expect(runMutation).toHaveBeenCalledWith( expect.anything(), + expect.objectContaining({ jobId: "registryArtifactBackupJobs:demo" }), ); }); - it("repairs multiple retry index misses for the same skill root with one index write", async () => { + it("marks multiple skill retries succeeded from matching version metadata", async () => { const jobs = [ makeSkillBackupJob("demo-1", "skillVersions:demo-1"), makeSkillBackupJob("demo-2", "skillVersions:demo-2"), @@ -1078,19 +952,10 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => { expect(result.stats.retryJobsSucceeded).toBe(2); expect(result.stats.retryJobsFailed).toBe(0); expect(registryBackupMocks.backupSkillVersionToObjectStorage).not.toHaveBeenCalled(); - expect(registryBackupMocks.repairSkillVersionBackupIndexes).toHaveBeenCalledOnce(); - expect(registryBackupMocks.repairSkillVersionBackupIndexes).toHaveBeenCalledWith( - expect.anything(), - [ - expect.objectContaining({ ownerHandle: "alice", slug: "demo-skill", version: "1.0.0" }), - expect.objectContaining({ ownerHandle: "alice", slug: "demo-skill", version: "1.1.0" }), - ], - expect.anything(), - expect.anything(), - ); + expect(registryBackupMocks.fetchSkillVersionBackupMeta).toHaveBeenCalledTimes(2); }); - it("repairs multiple retry index misses for the same package root with one index write", async () => { + it("marks package retries succeeded from matching version metadata", async () => { const jobs = [ makePackageBackupJob("demo-1", "packageReleases:demo-1"), makePackageBackupJob("demo-2", "packageReleases:demo-2"), @@ -1143,314 +1008,7 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => { expect(result.stats.retryJobsSucceeded).toBe(2); expect(result.stats.retryJobsFailed).toBe(0); expect(registryBackupMocks.backupPackageReleaseToObjectStorage).not.toHaveBeenCalled(); - expect(registryBackupMocks.repairPackageReleaseBackupIndexes).toHaveBeenCalledOnce(); - expect(registryBackupMocks.repairPackageReleaseBackupIndexes).toHaveBeenCalledWith( - expect.anything(), - [ - expect.objectContaining({ - ownerHandle: "alice", - normalizedName: "@openclaw/demo", - version: "1.0.0", - }), - expect.objectContaining({ - ownerHandle: "alice", - normalizedName: "@openclaw/demo", - version: "1.1.0", - }), - ], - expect.anything(), - expect.anything(), - ); - }); - - it("marks skill index retries succeeded when the version is already indexed", async () => { - const jobs = [ - makeSkillBackupJob("demo-1", "skillVersions:demo-1"), - makeSkillBackupJob("demo-2", "skillVersions:demo-2"), - ]; - const versions = new Map([ - ["skillVersions:demo-1", makeSkillVersion("skillVersions:demo-1", "skills:demo", "1.0.0")], - ["skillVersions:demo-2", makeSkillVersion("skillVersions:demo-2", "skills:demo", "1.1.0")], - ]); - const skill = { - ...makeSkill("skills:demo", "demo-skill"), - latestVersionId: "skillVersions:demo-2", - }; - const owner = { - _id: "users:owner", - handle: "alice", - deletedAt: undefined, - deactivatedAt: undefined, - }; - const runQuery = vi.fn(async (_ref, args) => { - if ("limit" in args) return jobs; - if (args.versionId) return versions.get(args.versionId) ?? null; - if (args.skillId === "skills:demo") return skill; - if (args.userId === "users:owner") return owner; - if ("staleAfterMs" in args) return { stale: 0, exhausted: 0 }; - throw new Error(`unexpected query ${JSON.stringify(args)}`); - }); - const versionIdsByVersion = new Map([ - ["1.0.0", "skillVersions:demo-1"], - ["1.1.0", "skillVersions:demo-2"], - ]); - registryBackupMocks.fetchSkillVersionBackupMeta.mockImplementation( - async (_context, _ownerHandle, _slug, version) => ({ - version, - restore: { versionId: versionIdsByVersion.get(version) }, - }), - ); - registryBackupMocks.fetchSkillBackupIndex.mockResolvedValueOnce({ - latest: { version: "1.1.0", versionId: "skillVersions:demo-2", isLatest: true }, - versions: [ - { version: "1.0.0", versionId: "skillVersions:demo-1", isLatest: false }, - { version: "1.1.0", versionId: "skillVersions:demo-2", isLatest: true }, - ], - }); - - const result = await processRegistryArtifactBackupRetriesInternalHandler( - { runQuery, runMutation: retryLeaseRunMutation() } as never, - {}, - ); - - expect(result.stats.retryJobsSucceeded).toBe(2); - expect(result.stats.retryJobsFailed).toBe(0); - expect(registryBackupMocks.repairSkillVersionBackupIndexes).not.toHaveBeenCalled(); - }); - - it("repairs skill index retries when the indexed version has stale latest state", async () => { - const jobs = [makeSkillBackupJob("demo", "skillVersions:demo")]; - const skill = { - ...makeSkill("skills:demo", "demo-skill"), - latestVersionId: "skillVersions:demo", - }; - const owner = { - _id: "users:owner", - handle: "alice", - deletedAt: undefined, - deactivatedAt: undefined, - }; - const runQuery = vi.fn(async (_ref, args) => { - if ("limit" in args) return jobs; - if (args.versionId) return makeSkillVersion("skillVersions:demo", "skills:demo", "1.0.0"); - if (args.skillId === "skills:demo") return skill; - if (args.userId === "users:owner") return owner; - if ("staleAfterMs" in args) return { stale: 0, exhausted: 0 }; - throw new Error(`unexpected query ${JSON.stringify(args)}`); - }); - registryBackupMocks.fetchSkillVersionBackupMeta.mockResolvedValue({ - version: "1.0.0", - restore: { versionId: "skillVersions:demo" }, - }); - registryBackupMocks.fetchSkillBackupIndex.mockResolvedValueOnce({ - latest: { version: "0.9.0", versionId: "skillVersions:old", isLatest: true }, - versions: [{ version: "1.0.0", versionId: "skillVersions:demo", isLatest: false }], - }); - - const result = await processRegistryArtifactBackupRetriesInternalHandler( - { runQuery, runMutation: retryLeaseRunMutation() } as never, - {}, - ); - - expect(result.stats.retryJobsSucceeded).toBe(1); - expect(result.stats.retryJobsFailed).toBe(0); - expect(registryBackupMocks.repairSkillVersionBackupIndexes).toHaveBeenCalledOnce(); - }); - - it("keeps indexed skill success marker failures isolated", async () => { - const jobs = [makeSkillBackupJob("demo", "skillVersions:demo")]; - const skill = { - ...makeSkill("skills:demo", "demo-skill"), - latestVersionId: "skillVersions:demo", - }; - const owner = { - _id: "users:owner", - handle: "alice", - deletedAt: undefined, - deactivatedAt: undefined, - }; - const runQuery = vi.fn(async (_ref, args) => { - if ("limit" in args) return jobs; - if (args.versionId) return makeSkillVersion("skillVersions:demo", "skills:demo", "1.0.0"); - if (args.skillId === "skills:demo") return skill; - if (args.userId === "users:owner") return owner; - if ("staleAfterMs" in args) return { stale: 0, exhausted: 0 }; - throw new Error(`unexpected query ${JSON.stringify(args)}`); - }); - const runMutation = vi.fn(async (_ref, args) => { - if (args && typeof args === "object" && "token" in args) { - return { acquired: true, released: true }; - } - if (args && typeof args === "object" && "jobId" in args) { - throw new Error("status patch failed"); - } - return undefined; - }); - registryBackupMocks.fetchSkillVersionBackupMeta.mockResolvedValue({ - version: "1.0.0", - restore: { versionId: "skillVersions:demo" }, - }); - registryBackupMocks.fetchSkillBackupIndex.mockResolvedValueOnce({ - latest: { version: "1.0.0", versionId: "skillVersions:demo", isLatest: true }, - versions: [{ version: "1.0.0", versionId: "skillVersions:demo", isLatest: true }], - }); - - const result = await processRegistryArtifactBackupRetriesInternalHandler( - { runQuery, runMutation } as never, - {}, - ); - - expect(result.stats.retryJobsSucceeded).toBe(0); - expect(result.stats.retryJobsFailed).toBe(1); - expect(registryBackupMocks.repairSkillVersionBackupIndexes).not.toHaveBeenCalled(); - expect(runMutation).toHaveBeenCalledTimes(4); - }); - - it("marks package index retries succeeded when the release is already indexed", async () => { - const jobs = [ - makePackageBackupJob("demo-1", "packageReleases:demo-1"), - makePackageBackupJob("demo-2", "packageReleases:demo-2"), - ]; - const releases = new Map([ - [ - "packageReleases:demo-1", - makePackageRelease("packageReleases:demo-1", "packages:demo", "1.0.0"), - ], - [ - "packageReleases:demo-2", - makePackageRelease("packageReleases:demo-2", "packages:demo", "1.1.0"), - ], - ]); - const pkg = { - ...makePackage("packages:demo", "@openclaw/demo"), - latestReleaseId: "packageReleases:demo-2", - }; - const owner = { - _id: "users:owner", - handle: "alice", - deletedAt: undefined, - deactivatedAt: undefined, - }; - const runQuery = vi.fn(async (_ref, args) => { - if ("limit" in args) return jobs; - if (args.releaseId) return releases.get(args.releaseId) ?? null; - if (args.packageId === "packages:demo") return pkg; - if (args.userId === "users:owner") return owner; - if ("staleAfterMs" in args) return { stale: 0, exhausted: 0 }; - throw new Error(`unexpected query ${JSON.stringify(args)}`); - }); - const releaseIdsByVersion = new Map([ - ["1.0.0", "packageReleases:demo-1"], - ["1.1.0", "packageReleases:demo-2"], - ]); - const shaByVersion = new Map([ - ["1.0.0", "sha:packageReleases:demo-1"], - ["1.1.0", "sha:packageReleases:demo-2"], - ]); - registryBackupMocks.fetchPackageReleaseBackupMeta.mockImplementation( - async (_context, _ownerHandle, _normalizedName, version) => ({ - restore: { releaseId: releaseIdsByVersion.get(version) }, - artifact: { sha256: shaByVersion.get(version) }, - }), - ); - registryBackupMocks.fetchPackageBackupIndex.mockResolvedValueOnce({ - latest: { version: "1.1.0", releaseId: "packageReleases:demo-2", isLatest: true }, - versions: [ - { version: "1.0.0", releaseId: "packageReleases:demo-1", isLatest: false }, - { version: "1.1.0", releaseId: "packageReleases:demo-2", isLatest: true }, - ], - }); - - const result = await processRegistryArtifactBackupRetriesInternalHandler( - { runQuery, runMutation: retryLeaseRunMutation() } as never, - {}, - ); - - expect(result.stats.retryJobsSucceeded).toBe(2); - expect(result.stats.retryJobsFailed).toBe(0); - expect(registryBackupMocks.repairPackageReleaseBackupIndexes).not.toHaveBeenCalled(); - }); - - it("marks skill index retries failed when the index lookup fails", async () => { - const jobs = [makeSkillBackupJob("demo", "skillVersions:demo")]; - const skill = makeSkill("skills:demo", "demo-skill"); - const owner = { - _id: "users:owner", - handle: "alice", - deletedAt: undefined, - deactivatedAt: undefined, - }; - const runQuery = vi.fn(async (_ref, args) => { - if ("limit" in args) return jobs; - if (args.versionId) return makeSkillVersion("skillVersions:demo", "skills:demo", "1.0.0"); - if (args.skillId === "skills:demo") return skill; - if (args.userId === "users:owner") return owner; - if ("staleAfterMs" in args) return { stale: 0, exhausted: 0 }; - throw new Error(`unexpected query ${JSON.stringify(args)}`); - }); - const runMutation = retryLeaseRunMutation(); - registryBackupMocks.fetchSkillVersionBackupMeta.mockResolvedValue({ - version: "1.0.0", - restore: { versionId: "skillVersions:demo" }, - }); - registryBackupMocks.fetchSkillBackupIndex.mockRejectedValueOnce(new Error("R2 index read")); - - const result = await processRegistryArtifactBackupRetriesInternalHandler( - { runQuery, runMutation } as never, - {}, - ); - - expect(result.stats.retryJobsSucceeded).toBe(0); - expect(result.stats.retryJobsFailed).toBe(1); - expect(registryBackupMocks.repairSkillVersionBackupIndexes).not.toHaveBeenCalled(); - expect(runMutation).toHaveBeenCalledWith( - expect.anything(), - expect.objectContaining({ - jobId: "registryArtifactBackupJobs:demo", - error: "R2 index read", - }), - ); - }); - - it("marks package index retries failed when the index lookup fails", async () => { - const jobs = [makePackageBackupJob("demo", "packageReleases:demo")]; - const pkg = makePackage("packages:demo", "@openclaw/demo"); - const owner = { - _id: "users:owner", - handle: "alice", - deletedAt: undefined, - deactivatedAt: undefined, - }; - const runQuery = vi.fn(async (_ref, args) => { - if ("limit" in args) return jobs; - if (args.releaseId) return makePackageRelease("packageReleases:demo", "packages:demo"); - if (args.packageId === "packages:demo") return pkg; - if (args.userId === "users:owner") return owner; - if ("staleAfterMs" in args) return { stale: 0, exhausted: 0 }; - throw new Error(`unexpected query ${JSON.stringify(args)}`); - }); - const runMutation = retryLeaseRunMutation(); - registryBackupMocks.fetchPackageReleaseBackupMeta.mockResolvedValue({ - restore: { releaseId: "packageReleases:demo" }, - artifact: { sha256: "sha:packageReleases:demo" }, - }); - registryBackupMocks.fetchPackageBackupIndex.mockRejectedValueOnce(new Error("R2 index read")); - - const result = await processRegistryArtifactBackupRetriesInternalHandler( - { runQuery, runMutation } as never, - {}, - ); - - expect(result.stats.retryJobsSucceeded).toBe(0); - expect(result.stats.retryJobsFailed).toBe(1); - expect(registryBackupMocks.repairPackageReleaseBackupIndexes).not.toHaveBeenCalled(); - expect(runMutation).toHaveBeenCalledWith( - expect.anything(), - expect.objectContaining({ - jobId: "registryArtifactBackupJobs:demo", - error: "R2 index read", - }), - ); + expect(registryBackupMocks.fetchPackageReleaseBackupMeta).toHaveBeenCalledTimes(2); }); it("processes different retry roots in parallel while keeping one root sequential", async () => { @@ -2009,91 +1567,6 @@ describe("registry artifact backup jobs", () => { }); }); - it("acquires a registry artifact backup index lease with an index-scoped key", async () => { - const now = 1_700_000_000_000; - const insert = vi.fn(); - const ctx = { - db: { - query: vi.fn(() => ({ - withIndex: vi.fn(() => ({ unique: vi.fn().mockResolvedValue(null) })), - })), - insert, - patch: vi.fn(), - }, - }; - - const result = await tryAcquireRegistryArtifactBackupIndexLeaseHandler(ctx as never, { - indexPath: "skills/alice/demo/_index.json", - now, - token: "index-token", - ttlMs: 60_000, - }); - - expect(result).toEqual({ acquired: true }); - expect(insert).toHaveBeenCalledWith("registryArtifactBackupSyncState", { - key: "index:skills/alice/demo/_index.json", - cursor: "index-token", - updatedAt: now, - }); - }); - - it("refuses a fresh registry artifact backup index lease", async () => { - const now = 1_700_000_000_000; - const existing = { - _id: "registryArtifactBackupSyncState:index", - key: "index:skills/alice/demo/_index.json", - cursor: "other-token", - updatedAt: now - 1_000, - }; - const patch = vi.fn(); - const ctx = { - db: { - query: vi.fn(() => ({ - withIndex: vi.fn(() => ({ unique: vi.fn().mockResolvedValue(existing) })), - })), - insert: vi.fn(), - patch, - }, - }; - - const result = await tryAcquireRegistryArtifactBackupIndexLeaseHandler(ctx as never, { - indexPath: "skills/alice/demo/_index.json", - now, - token: "index-token", - ttlMs: 60_000, - }); - - expect(result).toEqual({ acquired: false, holderUpdatedAt: existing.updatedAt }); - expect(patch).not.toHaveBeenCalled(); - }); - - it("releases only the matching registry artifact backup index lease token", async () => { - const now = 1_700_000_000_000; - const existing = { - _id: "registryArtifactBackupSyncState:index", - key: "index:skills/alice/demo/_index.json", - cursor: "index-token", - updatedAt: now - 1_000, - }; - const deleteDoc = vi.fn(); - const ctx = { - db: { - query: vi.fn(() => ({ - withIndex: vi.fn(() => ({ unique: vi.fn().mockResolvedValue(existing) })), - })), - delete: deleteDoc, - }, - }; - - const result = await releaseRegistryArtifactBackupIndexLeaseHandler(ctx as never, { - indexPath: "skills/alice/demo/_index.json", - token: "index-token", - }); - - expect(result).toEqual({ released: true }); - expect(deleteDoc).toHaveBeenCalledWith("registryArtifactBackupSyncState:index"); - }); - it("upserts package release backup failures into a retryable backlog", async () => { const now = 1_700_000_000_000; const existing = { diff --git a/convex/registryArtifactBackups.ts b/convex/registryArtifactBackups.ts index c409f03c..b044958e 100644 --- a/convex/registryArtifactBackups.ts +++ b/convex/registryArtifactBackups.ts @@ -12,7 +12,6 @@ const MAX_BATCH_SIZE = 200; const SYNC_STATE_KEY = "default"; const PACKAGE_SYNC_STATE_KEY = "packageReleases"; const RETRY_LEASE_KEY = "retryLease"; -const INDEX_LEASE_KEY_PREFIX = "index:"; const MAX_BACKUP_JOB_ERROR_LENGTH = 4000; const DEFAULT_BACKUP_HEALTH_SAMPLE_LIMIT = 500; const MAX_BACKUP_HEALTH_SAMPLE_LIMIT = 1000; @@ -21,8 +20,6 @@ const MAX_BACKUP_JOB_LIMIT = 500; const DEFAULT_BACKUP_JOB_REPAIR_ATTEMPTS = 16; const DEFAULT_RETRY_LEASE_TTL_MS = 20 * 60 * 1000; const MAX_RETRY_LEASE_TTL_MS = 60 * 60 * 1000; -const DEFAULT_INDEX_LEASE_TTL_MS = 5 * 60 * 1000; -const MAX_INDEX_LEASE_TTL_MS = 30 * 60 * 1000; type BackupPageItem = | { @@ -413,69 +410,6 @@ export const releaseRegistryArtifactBackupRetryLeaseInternal = internalMutation( handler: releaseRegistryArtifactBackupRetryLeaseHandler, }); -export async function tryAcquireRegistryArtifactBackupIndexLeaseHandler( - ctx: Pick, - args: { indexPath: string; now?: number; token: string; ttlMs?: number }, -) { - const now = args.now ?? Date.now(); - const ttlMs = clampInt(args.ttlMs ?? DEFAULT_INDEX_LEASE_TTL_MS, 1_000, MAX_INDEX_LEASE_TTL_MS); - const key = registryArtifactBackupIndexLeaseKey(args.indexPath); - const state = await ctx.db - .query("registryArtifactBackupSyncState") - .withIndex("by_key", (q) => q.eq("key", key)) - .unique(); - if (state?.cursor && state.updatedAt + ttlMs > now) { - return { acquired: false as const, holderUpdatedAt: state.updatedAt }; - } - - if (!state) { - await ctx.db.insert("registryArtifactBackupSyncState", { - key, - cursor: args.token, - updatedAt: now, - }); - return { acquired: true as const }; - } - - await ctx.db.patch(state._id, { - cursor: args.token, - updatedAt: now, - }); - return { acquired: true as const }; -} - -export const tryAcquireRegistryArtifactBackupIndexLeaseInternal = internalMutation({ - args: { - indexPath: v.string(), - now: v.optional(v.number()), - token: v.string(), - ttlMs: v.optional(v.number()), - }, - handler: tryAcquireRegistryArtifactBackupIndexLeaseHandler, -}); - -export async function releaseRegistryArtifactBackupIndexLeaseHandler( - ctx: Pick, - args: { indexPath: string; token: string }, -) { - const state = await ctx.db - .query("registryArtifactBackupSyncState") - .withIndex("by_key", (q) => q.eq("key", registryArtifactBackupIndexLeaseKey(args.indexPath))) - .unique(); - if (!state || state.cursor !== args.token) return { released: false as const }; - - await ctx.db.delete(state._id); - return { released: true as const }; -} - -export const releaseRegistryArtifactBackupIndexLeaseInternal = internalMutation({ - args: { - indexPath: v.string(), - token: v.string(), - }, - handler: releaseRegistryArtifactBackupIndexLeaseHandler, -}); - const registryArtifactBackupTargetKindValidator = v.union( v.literal("skillVersion"), v.literal("packageRelease"), @@ -724,10 +658,6 @@ function truncateBackupJobError(error: string | undefined) { return error.slice(0, MAX_BACKUP_JOB_ERROR_LENGTH); } -function registryArtifactBackupIndexLeaseKey(indexPath: string) { - return `${INDEX_LEASE_KEY_PREFIX}${indexPath}`; -} - function retryDelayMs(attempts: number) { const minutes = Math.min(60, 2 ** Math.min(attempts, 6)); return minutes * 60 * 1000; diff --git a/convex/registryArtifactBackupsNode.ts b/convex/registryArtifactBackupsNode.ts index d0e9a1cf..b19a0ea9 100644 --- a/convex/registryArtifactBackupsNode.ts +++ b/convex/registryArtifactBackupsNode.ts @@ -9,14 +9,10 @@ import { isPublicSkillDoc } from "./lib/globalStats"; import { backupPackageReleaseToObjectStorage, backupSkillVersionToObjectStorage, - fetchPackageBackupIndex, fetchPackageReleaseBackupMeta, - fetchSkillBackupIndex, fetchSkillVersionBackupMeta, getRegistryArtifactBackupContext, isRegistryArtifactBackupConfigured, - repairPackageReleaseBackupIndexes, - repairSkillVersionBackupIndexes, type RegistryArtifactBackupContext, } from "./lib/registryArtifactBackup"; @@ -32,9 +28,6 @@ const UNKNOWN_SKILL_ARTIFACT_BYTES = 50 * 1024 * 1024; const MAX_PARALLEL_RETRY_ARTIFACT_BYTES = UNKNOWN_PACKAGE_ARTIFACT_BYTES; const STALE_BACKUP_JOB_MS = 24 * 60 * 60 * 1000; const RETRY_LEASE_TTL_MS = 20 * 60 * 1000; -const INDEX_LEASE_TTL_MS = 5 * 60 * 1000; -const INDEX_LEASE_RETRY_DELAY_MS = 250; -const INDEX_LEASE_MAX_WAIT_MS = 30_000; type BackupPageItem = | { @@ -159,7 +152,7 @@ export const backupSkillForPublishInternal = internalAction({ if (args.versionId && !item) { return { skipped: true as const }; } - await backupSkillVersionWithIndexLease(ctx, item ?? args); + await backupSkillVersionToObjectStorage(ctx, item ?? args); return { skipped: false as const }; } catch (error) { if (args.versionId) { @@ -219,7 +212,7 @@ export const backupPackageForPublishInternal = internalAction({ if (!item) { return { skipped: true as const }; } - await backupPackageReleaseWithIndexLease(ctx, item); + await backupPackageReleaseToObjectStorage(ctx, item); return { skipped: false as const }; } catch (error) { await ctx.runMutation( @@ -314,7 +307,7 @@ export async function seedRegistryArtifactBackupsInternalHandler( } if (!dryRun) { - await backupSkillVersionWithIndexLease( + await backupSkillVersionToObjectStorage( ctx, { skillId: item.skillId, @@ -437,7 +430,7 @@ async function syncPackageReleaseBackups( continue; } if (!dryRun) { - await backupPackageReleaseWithIndexLease(ctx, item, context); + await backupPackageReleaseToObjectStorage(ctx, item, context); stats.packagesBackedUp += 1; } } catch (error) { @@ -527,10 +520,6 @@ type RetryJobWorkItem = estimatedBytes: number; }; -type ArtifactRetryJobWorkItem = - | Extract - | Extract; - type RetryJobGroup = { estimatedBytes: number; items: RetryJobWorkItem[]; @@ -598,8 +587,6 @@ async function processRetryJobGroup( group: RetryJobGroup, ) { const result = { processed: 0, succeeded: 0, failed: 0 }; - const packageIndexRepairs: Array> = []; - const skillIndexRepairs: Array> = []; for (const workItem of group.items) { result.processed += 1; try { @@ -610,17 +597,19 @@ async function processRetryJobGroup( result.succeeded += 1; } else if (workItem.kind === "packageRelease") { if (await hasMatchingPackageReleaseMeta(context, workItem.item)) { - packageIndexRepairs.push(workItem); + await markRetryJobSucceeded(ctx, workItem.job); + result.succeeded += 1; } else { - await backupPackageReleaseWithIndexLease(ctx, workItem.item, context); + await backupPackageReleaseToObjectStorage(ctx, workItem.item, context); await markRetryJobSucceeded(ctx, workItem.job); result.succeeded += 1; } } else { if (await hasMatchingSkillVersionMeta(context, workItem.item)) { - skillIndexRepairs.push(workItem); + await markRetryJobSucceeded(ctx, workItem.job); + result.succeeded += 1; } else { - await backupSkillVersionWithIndexLease(ctx, workItem.item, context); + await backupSkillVersionToObjectStorage(ctx, workItem.item, context); await markRetryJobSucceeded(ctx, workItem.job); result.succeeded += 1; } @@ -637,200 +626,9 @@ async function processRetryJobGroup( ); } } - await flushPackageIndexRepairs(ctx, context, packageIndexRepairs, result); - await flushSkillIndexRepairs(ctx, context, skillIndexRepairs, result); return result; } -async function flushPackageIndexRepairs( - ctx: ActionCtx, - context: RegistryArtifactBackupContext, - workItems: Array>, - result: { succeeded: number; failed: number }, -) { - if (workItems.length === 0) return; - let splitItems: { - indexed: Array>; - missing: Array>; - }; - try { - splitItems = await splitPackageIndexRepairItems(context, workItems); - } catch (error) { - result.failed += workItems.length; - await markRetryJobsFailed(ctx, workItems, error); - return; - } - - const { indexed, missing } = splitItems; - await markIndexedRetryJobsSucceeded(ctx, indexed, result); - if (missing.length === 0) return; - - try { - await repairPackageReleaseBackupIndexes( - ctx, - missing.map((workItem) => workItem.item), - context, - { - withIndexWrite: (indexPath, write) => - withRegistryArtifactBackupIndexLease(ctx, indexPath, write), - }, - ); - for (const workItem of missing) { - await markRetryJobSucceeded(ctx, workItem.job); - } - result.succeeded += missing.length; - } catch (error) { - result.failed += missing.length; - await markRetryJobsFailed(ctx, missing, error); - } -} - -async function flushSkillIndexRepairs( - ctx: ActionCtx, - context: RegistryArtifactBackupContext, - workItems: Array>, - result: { succeeded: number; failed: number }, -) { - if (workItems.length === 0) return; - let splitItems: { - indexed: Array>; - missing: Array>; - }; - try { - splitItems = await splitSkillIndexRepairItems(context, workItems); - } catch (error) { - result.failed += workItems.length; - await markRetryJobsFailed(ctx, workItems, error); - return; - } - - const { indexed, missing } = splitItems; - await markIndexedRetryJobsSucceeded(ctx, indexed, result); - if (missing.length === 0) return; - - try { - await repairSkillVersionBackupIndexes( - ctx, - missing.map((workItem) => workItem.item), - context, - { - withIndexWrite: (indexPath, write) => - withRegistryArtifactBackupIndexLease(ctx, indexPath, write), - }, - ); - for (const workItem of missing) { - await markRetryJobSucceeded(ctx, workItem.job); - } - result.succeeded += missing.length; - } catch (error) { - result.failed += missing.length; - await markRetryJobsFailed(ctx, missing, error); - } -} - -async function splitPackageIndexRepairItems( - context: RegistryArtifactBackupContext, - workItems: Array>, -) { - const first = workItems[0]; - const index = first - ? await fetchPackageBackupIndex(context, first.item.ownerHandle, first.item.normalizedName) - : null; - const indexed: Array> = []; - const missing: Array> = []; - for (const workItem of workItems) { - const present = packageIndexEntryMatchesRetry(index, workItem.item); - (present ? indexed : missing).push(workItem); - } - return { indexed, missing }; -} - -async function splitSkillIndexRepairItems( - context: RegistryArtifactBackupContext, - workItems: Array>, -) { - const first = workItems[0]; - const index = first - ? await fetchSkillBackupIndex(context, first.item.ownerHandle, first.item.slug) - : null; - const indexed: Array> = []; - const missing: Array> = []; - for (const workItem of workItems) { - const present = skillIndexEntryMatchesRetry(index, workItem.item); - (present ? indexed : missing).push(workItem); - } - return { indexed, missing }; -} - -function packageIndexEntryMatchesRetry( - index: Awaited>, - item: Extract["item"], -) { - const entry = index?.versions.find( - (candidate) => candidate.releaseId === item.releaseId && candidate.version === item.version, - ); - if (!entry) return false; - if (typeof item.isLatest !== "boolean") return true; - if (entry.isLatest !== item.isLatest) return false; - const latestMatches = Boolean( - index && index.latest.releaseId === item.releaseId && index.latest.version === item.version, - ); - return item.isLatest ? latestMatches : !latestMatches; -} - -function skillIndexEntryMatchesRetry( - index: Awaited>, - item: Extract["item"], -) { - const entry = index?.versions.find( - (candidate) => candidate.versionId === item.versionId && candidate.version === item.version, - ); - if (!entry) return false; - if (typeof item.isLatest !== "boolean") return true; - if (entry.isLatest !== item.isLatest) return false; - const latestMatches = Boolean( - index && index.latest.versionId === item.versionId && index.latest.version === item.version, - ); - return item.isLatest ? latestMatches : !latestMatches; -} - -async function markIndexedRetryJobsSucceeded( - ctx: ActionCtx, - workItems: ArtifactRetryJobWorkItem[], - result: { succeeded: number; failed: number }, -) { - for (const workItem of workItems) { - try { - await markRetryJobSucceeded(ctx, workItem.job); - result.succeeded += 1; - } catch (error) { - result.failed += 1; - try { - await markRetryJobsFailed(ctx, [workItem], error); - } catch (markFailedError) { - console.error("Registry artifact backup retry status update failed", markFailedError); - } - } - } -} - -async function markRetryJobsFailed( - ctx: ActionCtx, - workItems: ArtifactRetryJobWorkItem[], - error: unknown, -) { - for (const workItem of workItems) { - await ctx.runMutation( - internal.registryArtifactBackups.markRegistryArtifactBackupJobFailedInternal, - { - jobId: workItem.job._id, - error: errorMessage(error), - maxAttempts: MAX_RETRY_REPAIR_ATTEMPTS, - }, - ); - } -} - async function markRetryJobSucceeded(ctx: ActionCtx, job: Doc<"registryArtifactBackupJobs">) { await ctx.runMutation( internal.registryArtifactBackups.markRegistryArtifactBackupJobSucceededInternal, @@ -905,68 +703,6 @@ function chunkRetryJobGroups(groups: RetryJobGroup[]) { return chunks; } -async function backupSkillVersionWithIndexLease( - ctx: ActionCtx, - item: Parameters[1], - context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), -) { - await backupSkillVersionToObjectStorage(ctx, item, context, { - withIndexWrite: (indexPath, write) => - withRegistryArtifactBackupIndexLease(ctx, indexPath, write), - }); -} - -async function backupPackageReleaseWithIndexLease( - ctx: ActionCtx, - item: Parameters[1], - context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), -) { - await backupPackageReleaseToObjectStorage(ctx, item, context, { - withIndexWrite: (indexPath, write) => - withRegistryArtifactBackupIndexLease(ctx, indexPath, write), - }); -} - -async function withRegistryArtifactBackupIndexLease( - ctx: ActionCtx, - indexPath: string, - run: () => Promise, -) { - const token = `index-${Date.now()}-${Math.random().toString(36).slice(2)}`; - const deadline = Date.now() + INDEX_LEASE_MAX_WAIT_MS; - while (true) { - const lease = (await ctx.runMutation( - internal.registryArtifactBackups.tryAcquireRegistryArtifactBackupIndexLeaseInternal, - { - indexPath, - token, - ttlMs: INDEX_LEASE_TTL_MS, - }, - )) as { acquired: boolean }; - if (lease.acquired) break; - if (Date.now() >= deadline) { - throw new Error(`Registry artifact backup index ${indexPath} is busy`); - } - await sleep(INDEX_LEASE_RETRY_DELAY_MS); - } - - try { - return await run(); - } finally { - await ctx.runMutation( - internal.registryArtifactBackups.releaseRegistryArtifactBackupIndexLeaseInternal, - { - indexPath, - token, - }, - ); - } -} - -function sleep(ms: number) { - return new Promise((resolve) => setTimeout(resolve, ms)); -} - async function getPackageBackupItemForRelease( ctx: ActionCtx, releaseId: Id<"packageReleases">, diff --git a/convex/registryArtifactRestore.test.ts b/convex/registryArtifactRestore.test.ts index 6f8edb55..c48444db 100644 --- a/convex/registryArtifactRestore.test.ts +++ b/convex/registryArtifactRestore.test.ts @@ -2,7 +2,6 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { restoreSkillFromBackup } from "./registryArtifactRestore"; const registryBackupMocks = vi.hoisted(() => ({ - fetchSkillBackupIndex: vi.fn(), fetchSkillVersionBackupMeta: vi.fn(), getRegistryArtifactBackupContext: vi.fn(), isRegistryArtifactBackupConfigured: vi.fn(), @@ -54,6 +53,7 @@ describe("restoreSkillFromBackup", () => { ownerHandle: "alice", ownerUserId: "users:owner", slug: "demo-skill", + version: "1.0.0", }, ); @@ -62,13 +62,70 @@ describe("restoreSkillFromBackup", () => { status: "error", detail: "Existing skill is not public; restore blocked", }); - expect(registryBackupMocks.fetchSkillBackupIndex).not.toHaveBeenCalled(); + expect(registryBackupMocks.fetchSkillVersionBackupMeta).not.toHaveBeenCalled(); + }); + + it("requires an explicit version before any forced slug eviction", async () => { + const runMutation = vi.fn(); + const result = await restoreHandler( + { + runQuery: vi.fn().mockResolvedValueOnce({ _id: "users:admin", role: "admin" }), + runMutation, + } as never, + { + actorUserId: "users:admin", + ownerHandle: "alice", + ownerUserId: "users:owner", + slug: "demo-skill", + forceOverwriteSquatter: true, + }, + ); + + expect(result).toEqual({ + slug: "demo-skill", + status: "no_backup", + detail: "Restore requires an explicit backup version", + }); + expect(runMutation).not.toHaveBeenCalled(); + expect(registryBackupMocks.fetchSkillVersionBackupMeta).not.toHaveBeenCalled(); + }); + + it("validates the requested version before forced slug eviction", async () => { + registryBackupMocks.fetchSkillVersionBackupMeta.mockResolvedValueOnce(null); + const runMutation = vi.fn(); + const result = await restoreHandler( + { + runQuery: vi + .fn() + .mockResolvedValueOnce({ _id: "users:admin", role: "admin" }) + .mockResolvedValueOnce({ + _id: "skills:squatter", + ownerUserId: "users:other", + slug: "demo-skill", + softDeletedAt: undefined, + moderationStatus: "active", + }), + runMutation, + } as never, + { + actorUserId: "users:admin", + ownerHandle: "alice", + ownerUserId: "users:owner", + slug: "demo-skill", + version: "typo-version", + forceOverwriteSquatter: true, + }, + ); + + expect(result).toEqual({ + slug: "demo-skill", + status: "no_backup", + detail: "No version backup found", + }); + expect(runMutation).not.toHaveBeenCalled(); }); it("reactivates the same owner's soft-deleted skill row without republishing a duplicate version", async () => { - registryBackupMocks.fetchSkillBackupIndex.mockResolvedValueOnce({ - latest: { version: "1.0.0" }, - }); registryBackupMocks.fetchSkillVersionBackupMeta.mockResolvedValueOnce({ version: "1.0.0", displayName: "Demo Skill", @@ -114,6 +171,7 @@ describe("restoreSkillFromBackup", () => { ownerHandle: "alice", ownerUserId: "users:owner", slug: "demo-skill", + version: "1.0.0", }, ); @@ -148,9 +206,6 @@ describe("restoreSkillFromBackup", () => { }); it("fails restore when a manifest file is missing from backup storage", async () => { - registryBackupMocks.fetchSkillBackupIndex.mockResolvedValueOnce({ - latest: { version: "1.0.0" }, - }); registryBackupMocks.fetchSkillVersionBackupMeta.mockResolvedValueOnce({ version: "1.0.0", displayName: "Demo Skill", @@ -174,6 +229,7 @@ describe("restoreSkillFromBackup", () => { ownerHandle: "alice", ownerUserId: "users:owner", slug: "demo-skill", + version: "1.0.0", }, ); @@ -187,9 +243,6 @@ describe("restoreSkillFromBackup", () => { }); it("fails restore when a manifest file checksum does not match backup storage", async () => { - registryBackupMocks.fetchSkillBackupIndex.mockResolvedValueOnce({ - latest: { version: "1.0.0" }, - }); registryBackupMocks.fetchSkillVersionBackupMeta.mockResolvedValueOnce({ version: "1.0.0", displayName: "Demo Skill", @@ -215,6 +268,7 @@ describe("restoreSkillFromBackup", () => { ownerHandle: "alice", ownerUserId: "users:owner", slug: "demo-skill", + version: "1.0.0", }, ); diff --git a/convex/registryArtifactRestore.ts b/convex/registryArtifactRestore.ts index 422ee2a8..bfe9f6ee 100644 --- a/convex/registryArtifactRestore.ts +++ b/convex/registryArtifactRestore.ts @@ -8,7 +8,6 @@ import { assertAdmin } from "./lib/access"; import { guessContentTypeForPath } from "./lib/contentTypes"; import { isPublicSkillDoc } from "./lib/globalStats"; import { - fetchSkillBackupIndex, fetchSkillVersionBackupMeta, getRegistryArtifactBackupContext, isRegistryArtifactBackupConfigured, @@ -32,6 +31,19 @@ type BulkRestoreResult = { totalErrors: number; }; +type SkillBackupMeta = NonNullable>>; + +type VerifiedSkillBackup = { + meta: SkillBackupMeta; + files: Array<{ + path: string; + size: number; + sha256: string; + contentType: string; + content: Uint8Array; + }>; +}; + /** * Admin-only: restore a single skill from registry artifact backup. * Reads backed-up objects and re-creates the skill in the database. @@ -42,6 +54,7 @@ export const restoreSkillFromBackup = internalAction({ ownerHandle: v.string(), ownerUserId: v.id("users"), slug: v.string(), + version: v.optional(v.string()), forceOverwriteSquatter: v.optional(v.boolean()), }, handler: async (ctx, args): Promise => { @@ -59,6 +72,80 @@ export const restoreSkillFromBackup = internalAction({ } const backupContext = getRegistryArtifactBackupContext(); + if (!args.version) { + return { + slug: args.slug, + status: "no_backup", + detail: "Restore requires an explicit backup version", + }; + } + + let verifiedBackup: VerifiedSkillBackup | null = null; + const loadVerifiedBackup = async (): Promise => { + if (verifiedBackup) return verifiedBackup; + const meta = await fetchSkillVersionBackupMeta( + backupContext, + args.ownerHandle, + args.slug, + args.version!, + ); + if (!meta) { + return { slug: args.slug, status: "no_backup", detail: "No version backup found" }; + } + + const backupFiles = meta.metadata.files; + if (backupFiles.length === 0) { + return { slug: args.slug, status: "no_backup", detail: "Backup has no files" }; + } + + const owner = normalizeOwner(args.ownerHandle); + const files: VerifiedSkillBackup["files"] = []; + for (const file of backupFiles) { + if (!validateFilePath(file.path)) { + return { slug: args.slug, status: "error", detail: "Backup contains unsafe file path" }; + } + const fileContent = await readRegistryArtifactBackupObject( + backupContext, + `${backupContext.skillsRoot}/${owner}/${args.slug}/${encodeBackupPathSegment( + meta.version, + )}/${file.path}`, + ); + if (!fileContent) { + return { + slug: args.slug, + status: "error", + detail: `Backup missing file ${file.path}`, + }; + } + if (fileContent.byteLength !== file.size) { + return { + slug: args.slug, + status: "error", + detail: `Backup file size mismatch for ${file.path}`, + }; + } + + const sha256 = await sha256Hex(fileContent); + if (sha256 !== file.sha256) { + return { + slug: args.slug, + status: "error", + detail: `Backup file checksum mismatch for ${file.path}`, + }; + } + + files.push({ + path: file.path, + size: fileContent.byteLength, + sha256, + contentType: file.contentType ?? guessContentTypeForPath(file.path), + content: fileContent, + }); + } + + verifiedBackup = { meta, files }; + return verifiedBackup; + }; // Check if skill already exists in the DB const existingSkill = (await ctx.runQuery( @@ -97,6 +184,8 @@ export const restoreSkillFromBackup = internalAction({ detail: `Slug occupied by another user. Set forceOverwriteSquatter=true to reclaim.`, }; } else { + const backup = await loadVerifiedBackup(); + if ("status" in backup) return backup; // Free the slug in-transaction by renaming the squatter, then enqueue cleanup. await ctx.runMutation( internal.registryArtifactRestoreMutations.evictSquatterSkillForRestoreInternal, @@ -109,25 +198,9 @@ export const restoreSkillFromBackup = internalAction({ } } - const index = await fetchSkillBackupIndex(backupContext, args.ownerHandle, args.slug); - if (!index?.latest?.version) { - return { slug: args.slug, status: "no_backup", detail: "No backup index found" }; - } - - const meta = await fetchSkillVersionBackupMeta( - backupContext, - args.ownerHandle, - args.slug, - index.latest.version, - ); - if (!meta) { - return { slug: args.slug, status: "no_backup", detail: "No version backup found" }; - } - - const backupFiles = meta.metadata.files; - if (backupFiles.length === 0) { - return { slug: args.slug, status: "no_backup", detail: "Backup has no files" }; - } + const backup = verifiedBackup ?? (await loadVerifiedBackup()); + if ("status" in backup) return backup; + const { meta } = backup; // Download and store each file in Convex storage const storedFiles: Array<{ @@ -138,47 +211,16 @@ export const restoreSkillFromBackup = internalAction({ contentType: string; }> = []; - const owner = normalizeOwner(args.ownerHandle); - for (const file of backupFiles) { - if (!validateFilePath(file.path)) { - return { slug: args.slug, status: "error", detail: "Backup contains unsafe file path" }; - } - const fileContent = await readRegistryArtifactBackupObject( - backupContext, - `${backupContext.skillsRoot}/${owner}/${args.slug}/${encodeBackupPathSegment( - meta.version, - )}/${file.path}`, - ); - if (!fileContent) { - return { slug: args.slug, status: "error", detail: `Backup missing file ${file.path}` }; - } - if (fileContent.byteLength !== file.size) { - return { - slug: args.slug, - status: "error", - detail: `Backup file size mismatch for ${file.path}`, - }; - } - - const sha256 = await sha256Hex(fileContent); - if (sha256 !== file.sha256) { - return { - slug: args.slug, - status: "error", - detail: `Backup file checksum mismatch for ${file.path}`, - }; - } - - const contentType = file.contentType ?? guessContentTypeForPath(file.path); - const blob = new Blob([Buffer.from(fileContent)], { type: contentType }); + for (const file of backup.files) { + const blob = new Blob([Buffer.from(file.content)], { type: file.contentType }); const storageId = await ctx.storage.store(blob); storedFiles.push({ path: file.path, - size: fileContent.byteLength, + size: file.size, storageId, - sha256, - contentType, + sha256: file.sha256, + contentType: file.contentType, }); } @@ -251,6 +293,7 @@ export const restoreUserSkillsFromBackup = internalAction({ ownerHandle: v.string(), ownerUserId: v.id("users"), slugs: v.array(v.string()), + versionsBySlug: v.optional(v.record(v.string(), v.string())), forceOverwriteSquatter: v.optional(v.boolean()), }, handler: async (ctx, args): Promise => { @@ -266,6 +309,7 @@ export const restoreUserSkillsFromBackup = internalAction({ ownerHandle: args.ownerHandle, ownerUserId: args.ownerUserId, slug, + version: args.versionsBySlug?.[slug], forceOverwriteSquatter: args.forceOverwriteSquatter, })) as RestoreResult;