From 6ae7cb5345fa01cfca924a39fb883a7d2ff86b0c Mon Sep 17 00:00:00 2001 From: Patrick Erichsen Date: Sun, 14 Jun 2026 18:54:22 -0700 Subject: [PATCH] fix: bulk repair registry backup indexes --- convex/lib/registryArtifactBackup.ts | 66 +++++++++--- convex/registryArtifactBackups.test.ts | 144 +++++++++++++++++++++++-- convex/registryArtifactBackupsNode.ts | 87 +++++++++++++-- 3 files changed, 271 insertions(+), 26 deletions(-) diff --git a/convex/lib/registryArtifactBackup.ts b/convex/lib/registryArtifactBackup.ts index e035a0f6..ad2d5eea 100644 --- a/convex/lib/registryArtifactBackup.ts +++ b/convex/lib/registryArtifactBackup.ts @@ -195,12 +195,29 @@ export async function repairSkillVersionBackupIndex( params: SkillBackupParams & { root?: string }, context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), ) { - const planned = buildSkillVersionBackupManifest({ - root: params.root ?? context.skillsRoot, - ...params, - }); - await putMergedJsonIndex(context, planned.indexPath, (existingIndex: SkillIndexFile | null) => - buildSkillIndexFile(planned, existingIndex), + await repairSkillVersionBackupIndexes(_ctx, [params], context); +} + +export async function repairSkillVersionBackupIndexes( + _ctx: Pick, + params: Array, + context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), +) { + 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 putMergedJsonIndex(context, indexPath, (existingIndex: SkillIndexFile | null) => + rest.reduce( + (nextIndex, plannedItem) => buildSkillIndexFile(plannedItem, nextIndex), + buildSkillIndexFile(first, existingIndex), + ), ); } @@ -209,12 +226,29 @@ export async function repairPackageReleaseBackupIndex( params: PackageBackupParams & { root?: string }, context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), ) { - const planned = buildPackageReleaseBackupManifest({ - root: params.root ?? context.packagesRoot, - ...params, - }); - await putMergedJsonIndex(context, planned.indexPath, (existingIndex: PackageIndexFile | null) => - buildPackageIndexFile(planned, existingIndex), + await repairPackageReleaseBackupIndexes(_ctx, [params], context); +} + +export async function repairPackageReleaseBackupIndexes( + _ctx: Pick, + params: Array, + context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(), +) { + 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 putMergedJsonIndex(context, indexPath, (existingIndex: PackageIndexFile | null) => + rest.reduce( + (nextIndex, plannedItem) => buildPackageIndexFile(plannedItem, nextIndex), + buildPackageIndexFile(first, existingIndex), + ), ); } @@ -411,6 +445,14 @@ function buildSkillIndexFile( }; } +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); diff --git a/convex/registryArtifactBackups.test.ts b/convex/registryArtifactBackups.test.ts index df2c1fa2..5c577eec 100644 --- a/convex/registryArtifactBackups.test.ts +++ b/convex/registryArtifactBackups.test.ts @@ -22,7 +22,9 @@ const registryBackupMocks = vi.hoisted(() => ({ getRegistryArtifactBackupContext: vi.fn(), isRegistryArtifactBackupConfigured: vi.fn(), repairPackageReleaseBackupIndex: vi.fn(), + repairPackageReleaseBackupIndexes: vi.fn(), repairSkillVersionBackupIndex: vi.fn(), + repairSkillVersionBackupIndexes: vi.fn(), })); vi.mock("./lib/registryArtifactBackup", () => registryBackupMocks); @@ -801,13 +803,141 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => { expect(result.stats.retryJobsSucceeded).toBe(1); expect(registryBackupMocks.backupSkillVersionToObjectStorage).not.toHaveBeenCalled(); - expect(registryBackupMocks.repairSkillVersionBackupIndex).toHaveBeenCalledWith( + expect(registryBackupMocks.repairSkillVersionBackupIndexes).toHaveBeenCalledWith( expect.anything(), - expect.objectContaining({ - slug: "demo-skill", - version: "1.0.0", - ownerHandle: "alice", + [ + expect.objectContaining({ + slug: "demo-skill", + version: "1.0.0", + ownerHandle: "alice", + }), + ], + expect.anything(), + ); + }); + + it("repairs multiple retry index misses for the same skill root with one index write", 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"); + 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) }, }), + ); + + const result = await processRegistryArtifactBackupRetriesInternalHandler( + { runQuery, runMutation: vi.fn() } as never, + {}, + ); + + 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(), + ); + }); + + it("repairs multiple retry index misses for the same package root with one index write", 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"); + 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) }, + }), + ); + + const result = await processRegistryArtifactBackupRetriesInternalHandler( + { runQuery, runMutation: vi.fn() } as never, + {}, + ); + + 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(), ); }); @@ -1063,11 +1193,11 @@ function makePackageBackupJob(suffix: string, packageReleaseId: string) { }; } -function makePackageRelease(id: string, packageId: string) { +function makePackageRelease(id: string, packageId: string, version = "1.0.0") { return { _id: id, packageId, - version: "1.0.0", + version, createdAt: 1, files: [], clawpackStorageId: `storage:${id}`, diff --git a/convex/registryArtifactBackupsNode.ts b/convex/registryArtifactBackupsNode.ts index 52a9928d..45ae1639 100644 --- a/convex/registryArtifactBackupsNode.ts +++ b/convex/registryArtifactBackupsNode.ts @@ -13,8 +13,8 @@ import { fetchSkillVersionBackupMeta, getRegistryArtifactBackupContext, isRegistryArtifactBackupConfigured, - repairPackageReleaseBackupIndex, - repairSkillVersionBackupIndex, + repairPackageReleaseBackupIndexes, + repairSkillVersionBackupIndexes, type RegistryArtifactBackupContext, } from "./lib/registryArtifactBackup"; @@ -514,6 +514,10 @@ type RetryJobWorkItem = estimatedBytes: number; }; +type ArtifactRetryJobWorkItem = + | Extract + | Extract; + type RetryJobGroup = { estimatedBytes: number; items: RetryJobWorkItem[]; @@ -581,6 +585,8 @@ 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 { @@ -588,22 +594,24 @@ async function processRetryJobGroup( throw workItem.error; } else if (workItem.kind === "missing") { await markRetryJobSucceeded(ctx, workItem.job); + result.succeeded += 1; } else if (workItem.kind === "packageRelease") { if (await hasMatchingPackageReleaseMeta(context, workItem.item)) { - await repairPackageReleaseBackupIndex(ctx, workItem.item, context); + packageIndexRepairs.push(workItem); } else { await backupPackageReleaseToObjectStorage(ctx, workItem.item, context); + await markRetryJobSucceeded(ctx, workItem.job); + result.succeeded += 1; } - await markRetryJobSucceeded(ctx, workItem.job); } else { if (await hasMatchingSkillVersionMeta(context, workItem.item)) { - await repairSkillVersionBackupIndex(ctx, workItem.item, context); + skillIndexRepairs.push(workItem); } else { await backupSkillVersionToObjectStorage(ctx, workItem.item, context); + await markRetryJobSucceeded(ctx, workItem.job); + result.succeeded += 1; } - await markRetryJobSucceeded(ctx, workItem.job); } - result.succeeded += 1; } catch (error) { result.failed += 1; await ctx.runMutation( @@ -616,9 +624,74 @@ 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; + try { + await repairPackageReleaseBackupIndexes( + ctx, + workItems.map((workItem) => workItem.item), + context, + ); + for (const workItem of workItems) { + await markRetryJobSucceeded(ctx, workItem.job); + } + result.succeeded += workItems.length; + } catch (error) { + result.failed += workItems.length; + await markRetryJobsFailed(ctx, workItems, error); + } +} + +async function flushSkillIndexRepairs( + ctx: ActionCtx, + context: RegistryArtifactBackupContext, + workItems: Array>, + result: { succeeded: number; failed: number }, +) { + if (workItems.length === 0) return; + try { + await repairSkillVersionBackupIndexes( + ctx, + workItems.map((workItem) => workItem.item), + context, + ); + for (const workItem of workItems) { + await markRetryJobSucceeded(ctx, workItem.job); + } + result.succeeded += workItems.length; + } catch (error) { + result.failed += workItems.length; + await markRetryJobsFailed(ctx, workItems, error); + } +} + +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,