fix: bulk repair registry backup indexes

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<ActionCtx, "storage">,
+  params: Array<SkillBackupParams & { root?: string }>,
+  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<ActionCtx, "storage">,
+  params: Array<PackageBackupParams & { root?: string }>,
+  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);

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}`,

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<RetryJobWorkItem, { kind: "packageRelease" }>
+  | Extract<RetryJobWorkItem, { kind: "skillVersion" }>;
+
 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<Extract<RetryJobWorkItem, { kind: "packageRelease" }>> = [];
+  const skillIndexRepairs: Array<Extract<RetryJobWorkItem, { kind: "skillVersion" }>> = [];
   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<Extract<RetryJobWorkItem, { kind: "packageRelease" }>>,
+  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<Extract<RetryJobWorkItem, { kind: "skillVersion" }>>,
+  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,