admin/clawhub
1
0
Fork 0
mirror of https://github.com/openclaw/clawhub.git synced 2026-08-14 08:52:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: serialize registry backup index writes

Browse Source
This commit is contained in:
Patrick Erichsen
2026-06-14 20:37:34 -07:00
committed by GitHub
parent ee9c6c1412
commit b0c8fe2a01
4 changed files with 502 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
convex/lib/registryArtifactBackup.ts
+53 -16
View File
@@ -64,6 +64,10 @@ type PackageBackupParams = {
files: Array<{ path: string; size: number; sha256: string }>;
};
type IndexWriteOptions = {
withIndexWrite?: <T>(indexPath: string, write: () => Promise<T>) => Promise<T>;
};
type VersionIndexEntry = {
version: string;
isLatest?: boolean;
@@ -151,6 +155,7 @@ export async function backupSkillVersionToObjectStorage(
ctx: Pick<ActionCtx, "storage">,
params: SkillBackupParams & { root?: string },
context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(),
options: IndexWriteOptions = {},
) {
const planned = buildSkillVersionBackupManifest({
root: params.root ?? context.skillsRoot,
@@ -165,8 +170,11 @@ export async function backupSkillVersionToObjectStorage(
}
await putJsonObject(context, planned.metaPath, planned.meta);
await putMergedJsonIndex(context, planned.indexPath, (existingIndex: SkillIndexFile | null) =>
buildSkillIndexFile(planned, existingIndex),
await writeMergedJsonIndex(
context,
planned.indexPath,
(existingIndex: SkillIndexFile | null) => buildSkillIndexFile(planned, existingIndex),
options,
);
}
@@ -174,6 +182,7 @@ export async function backupPackageReleaseToObjectStorage(
ctx: Pick<ActionCtx, "storage">,
params: PackageBackupParams & { artifactStorageId: Id<"_storage">; root?: string },
context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(),
options: IndexWriteOptions = {},
) {
const planned = buildPackageReleaseBackupManifest({
root: params.root ?? context.packagesRoot,
@@ -185,8 +194,11 @@ export async function backupPackageReleaseToObjectStorage(
});
await putJsonObject(context, planned.metaPath, planned.meta);
await putMergedJsonIndex(context, planned.indexPath, (existingIndex: PackageIndexFile | null) =>
buildPackageIndexFile(planned, existingIndex),
await writeMergedJsonIndex(
context,
planned.indexPath,
(existingIndex: PackageIndexFile | null) => buildPackageIndexFile(planned, existingIndex),
options,
);
}
@@ -194,14 +206,16 @@ export async function repairSkillVersionBackupIndex(
_ctx: Pick<ActionCtx, "storage">,
params: SkillBackupParams & { root?: string },
context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(),
options: IndexWriteOptions = {},
) {
await repairSkillVersionBackupIndexes(_ctx, [params], context);
await repairSkillVersionBackupIndexes(_ctx, [params], context, options);
}
export async function repairSkillVersionBackupIndexes(
_ctx: Pick<ActionCtx, "storage">,
params: Array<SkillBackupParams & { root?: string }>,
context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(),
options: IndexWriteOptions = {},
) {
if (params.length === 0) return;
const planned = params.map((item) =>
@@ -213,11 +227,15 @@ export async function repairSkillVersionBackupIndexes(
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),
),
await writeMergedJsonIndex(
context,
indexPath,
(existingIndex: SkillIndexFile | null) =>
rest.reduce(
(nextIndex, plannedItem) => buildSkillIndexFile(plannedItem, nextIndex),
buildSkillIndexFile(first, existingIndex),
),
options,
);
}
@@ -225,14 +243,16 @@ export async function repairPackageReleaseBackupIndex(
_ctx: Pick<ActionCtx, "storage">,
params: PackageBackupParams & { root?: string },
context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(),
options: IndexWriteOptions = {},
) {
await repairPackageReleaseBackupIndexes(_ctx, [params], context);
await repairPackageReleaseBackupIndexes(_ctx, [params], context, options);
}
export async function repairPackageReleaseBackupIndexes(
_ctx: Pick<ActionCtx, "storage">,
params: Array<PackageBackupParams & { root?: string }>,
context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(),
options: IndexWriteOptions = {},
) {
if (params.length === 0) return;
const planned = params.map((item) =>
@@ -244,11 +264,15 @@ export async function repairPackageReleaseBackupIndexes(
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),
),
await writeMergedJsonIndex(
context,
indexPath,
(existingIndex: PackageIndexFile | null) =>
rest.reduce(
(nextIndex, plannedItem) => buildPackageIndexFile(plannedItem, nextIndex),
buildPackageIndexFile(first, existingIndex),
),
options,
);
}
@@ -265,6 +289,19 @@ export async function fetchSkillVersionBackupMeta(
return getJsonObject<ReturnType<typeof buildSkillVersionBackupManifest>["meta"]>(context, path);
}
async function writeMergedJsonIndex<T>(
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,
convex/registryArtifactBackups.test.ts
+300 -45
View File
@@ -7,6 +7,8 @@ import {
getRegistryArtifactBackupPageInternal,
getPackageRegistryArtifactBackupPageInternal,
releaseRegistryArtifactBackupRetryLeaseHandler,
releaseRegistryArtifactBackupIndexLeaseHandler,
tryAcquireRegistryArtifactBackupIndexLeaseHandler,
tryAcquireRegistryArtifactBackupRetryLeaseHandler,
} from "./registryArtifactBackups";
import {
@@ -16,20 +18,38 @@ import {
seedRegistryArtifactBackupsInternalHandler,
} from "./registryArtifactBackupsNode";
const registryBackupMocks = vi.hoisted(() => ({
backupPackageReleaseToObjectStorage: vi.fn(),
backupSkillVersionToObjectStorage: vi.fn(),
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(),
}));
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(),
};
});
vi.mock("./lib/registryArtifactBackup", () => registryBackupMocks);
@@ -50,7 +70,7 @@ const backupPackageForPublishHandler = (
beforeEach(() => {
vi.clearAllMocks();
registryBackupMocks.getRegistryArtifactBackupContext.mockReturnValue({
const backupContext = {
endpoint: "https://account.r2.cloudflarestorage.com",
bucket: "clawhub-registry-backup",
accessKeyId: "access-key",
@@ -58,12 +78,99 @@ beforeEach(() => {
region: "auto",
skillsRoot: "skills",
packagesRoot: "packages",
});
};
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<void>) => Promise<void>;
},
) => {
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<void>) => Promise<void>;
},
) => {
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<void>) => Promise<void>;
},
) => {
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<void>) => Promise<void>;
},
) => {
const first = params[0];
if (!first) return;
const manifest = registryBackupMocks.buildPackageReleaseBackupManifest({
root: first.root ?? context.packagesRoot,
...first,
});
await options?.withIndexWrite?.(manifest.indexPath, async () => undefined);
},
);
});
function retryLeaseRunMutation() {
return vi.fn().mockResolvedValueOnce({ acquired: true });
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;
});
}
describe("publish-time registry artifact backups", () => {
@@ -95,17 +202,20 @@ describe("publish-time registry artifact backups", () => {
deactivatedAt: undefined,
});
await backupSkillForPublishHandler({ runQuery, runMutation: vi.fn() } as never, {
skillId: "skills:demo",
versionId: "skillVersions:demo-1",
slug: "stale-slug",
version: "1.0.0",
isLatest: true,
displayName: "Stale Name",
ownerHandle: "stale-owner",
files: [],
publishedAt: 1,
});
await backupSkillForPublishHandler(
{ runQuery, runMutation: retryLeaseRunMutation() } as never,
{
skillId: "skills:demo",
versionId: "skillVersions:demo-1",
slug: "stale-slug",
version: "1.0.0",
isLatest: true,
displayName: "Stale Name",
ownerHandle: "stale-owner",
files: [],
publishedAt: 1,
},
);
expect(registryBackupMocks.backupSkillVersionToObjectStorage).toHaveBeenCalledWith(
expect.anything(),
@@ -115,6 +225,8 @@ describe("publish-time registry artifact backups", () => {
ownerHandle: "alice",
isLatest: false,
}),
expect.anything(),
expect.anything(),
);
});
@@ -151,20 +263,23 @@ describe("publish-time registry artifact backups", () => {
deactivatedAt: undefined,
});
await backupPackageForPublishHandler({ runQuery, runMutation: vi.fn() } as never, {
ownerHandle: "stale-owner",
packageId: "packages:demo",
releaseId: "packageReleases:demo-1",
packageName: "@openclaw/stale",
normalizedName: "@openclaw/stale",
displayName: "Stale Package",
family: "code-plugin",
version: "1.0.0",
isLatest: true,
publishedAt: 1,
artifactStorageId: "storage:artifact",
files: [],
});
await backupPackageForPublishHandler(
{ runQuery, runMutation: retryLeaseRunMutation() } as never,
{
ownerHandle: "stale-owner",
packageId: "packages:demo",
releaseId: "packageReleases:demo-1",
packageName: "@openclaw/stale",
normalizedName: "@openclaw/stale",
displayName: "Stale Package",
family: "code-plugin",
version: "1.0.0",
isLatest: true,
publishedAt: 1,
artifactStorageId: "storage:artifact",
files: [],
},
);
expect(registryBackupMocks.backupPackageReleaseToObjectStorage).toHaveBeenCalledWith(
expect.anything(),
@@ -175,6 +290,8 @@ describe("publish-time registry artifact backups", () => {
displayName: "Current Package",
isLatest: false,
}),
expect.anything(),
expect.anything(),
);
});
});
@@ -652,6 +769,50 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => {
expect(runQuery.mock.calls[0]?.[1]).toMatchObject({ ignoreNextRunAt: true });
});
it("serializes retry artifact backups with a per-index lease", 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 = retryLeaseRunMutation();
registryBackupMocks.fetchSkillVersionBackupMeta.mockResolvedValue(null);
const result = await processRegistryArtifactBackupRetriesInternalHandler(
{ runQuery, runMutation } as never,
{},
);
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",
}),
]);
});
it("drains retry jobs without scanning the historical registry", async () => {
const dueJob = {
_id: "registryArtifactBackupJobs:demo",
@@ -867,6 +1028,7 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => {
}),
],
expect.anything(),
expect.anything(),
);
});
@@ -924,6 +1086,7 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => {
expect.objectContaining({ ownerHandle: "alice", slug: "demo-skill", version: "1.1.0" }),
],
expect.anything(),
expect.anything(),
);
});
@@ -996,6 +1159,7 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => {
}),
],
expect.anything(),
expect.anything(),
);
});
@@ -1113,9 +1277,15 @@ describe("processRegistryArtifactBackupRetriesInternalHandler", () => {
if ("staleAfterMs" in args) return { stale: 0, exhausted: 0 };
throw new Error(`unexpected query ${JSON.stringify(args)}`);
});
const runMutation = retryLeaseRunMutation().mockRejectedValueOnce(
new Error("status patch failed"),
);
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" },
@@ -1839,6 +2009,91 @@ 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 = {
convex/registryArtifactBackups.ts
+70
View File
@@ -12,6 +12,7 @@ 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;
@@ -20,6 +21,8 @@ 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 =
| {
@@ -410,6 +413,69 @@ export const releaseRegistryArtifactBackupRetryLeaseInternal = internalMutation(
handler: releaseRegistryArtifactBackupRetryLeaseHandler,
});
export async function tryAcquireRegistryArtifactBackupIndexLeaseHandler(
ctx: Pick<MutationCtx, "db">,
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<MutationCtx, "db">,
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"),
@@ -658,6 +724,10 @@ 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;
convex/registryArtifactBackupsNode.ts
+79 -6
View File
@@ -32,6 +32,9 @@ 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 =
| {
@@ -156,7 +159,7 @@ export const backupSkillForPublishInternal = internalAction({
if (args.versionId && !item) {
return { skipped: true as const };
}
await backupSkillVersionToObjectStorage(ctx, item ?? args);
await backupSkillVersionWithIndexLease(ctx, item ?? args);
return { skipped: false as const };
} catch (error) {
if (args.versionId) {
@@ -216,7 +219,7 @@ export const backupPackageForPublishInternal = internalAction({
if (!item) {
return { skipped: true as const };
}
await backupPackageReleaseToObjectStorage(ctx, item);
await backupPackageReleaseWithIndexLease(ctx, item);
return { skipped: false as const };
} catch (error) {
await ctx.runMutation(
@@ -311,7 +314,7 @@ export async function seedRegistryArtifactBackupsInternalHandler(
}
if (!dryRun) {
await backupSkillVersionToObjectStorage(
await backupSkillVersionWithIndexLease(
ctx,
{
skillId: item.skillId,
@@ -434,7 +437,7 @@ async function syncPackageReleaseBackups(
continue;
}
if (!dryRun) {
await backupPackageReleaseToObjectStorage(ctx, item, context);
await backupPackageReleaseWithIndexLease(ctx, item, context);
stats.packagesBackedUp += 1;
}
} catch (error) {
@@ -609,7 +612,7 @@ async function processRetryJobGroup(
if (await hasMatchingPackageReleaseMeta(context, workItem.item)) {
packageIndexRepairs.push(workItem);
} else {
await backupPackageReleaseToObjectStorage(ctx, workItem.item, context);
await backupPackageReleaseWithIndexLease(ctx, workItem.item, context);
await markRetryJobSucceeded(ctx, workItem.job);
result.succeeded += 1;
}
@@ -617,7 +620,7 @@ async function processRetryJobGroup(
if (await hasMatchingSkillVersionMeta(context, workItem.item)) {
skillIndexRepairs.push(workItem);
} else {
await backupSkillVersionToObjectStorage(ctx, workItem.item, context);
await backupSkillVersionWithIndexLease(ctx, workItem.item, context);
await markRetryJobSucceeded(ctx, workItem.job);
result.succeeded += 1;
}
@@ -667,6 +670,10 @@ async function flushPackageIndexRepairs(
ctx,
missing.map((workItem) => workItem.item),
context,
{
withIndexWrite: (indexPath, write) =>
withRegistryArtifactBackupIndexLease(ctx, indexPath, write),
},
);
for (const workItem of missing) {
await markRetryJobSucceeded(ctx, workItem.job);
@@ -706,6 +713,10 @@ async function flushSkillIndexRepairs(
ctx,
missing.map((workItem) => workItem.item),
context,
{
withIndexWrite: (indexPath, write) =>
withRegistryArtifactBackupIndexLease(ctx, indexPath, write),
},
);
for (const workItem of missing) {
await markRetryJobSucceeded(ctx, workItem.job);
@@ -894,6 +905,68 @@ function chunkRetryJobGroups(groups: RetryJobGroup[]) {
return chunks;
}
async function backupSkillVersionWithIndexLease(
ctx: ActionCtx,
item: Parameters<typeof backupSkillVersionToObjectStorage>[1],
context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(),
) {
await backupSkillVersionToObjectStorage(ctx, item, context, {
withIndexWrite: (indexPath, write) =>
withRegistryArtifactBackupIndexLease(ctx, indexPath, write),
});
}
async function backupPackageReleaseWithIndexLease(
ctx: ActionCtx,
item: Parameters<typeof backupPackageReleaseToObjectStorage>[1],
context: RegistryArtifactBackupContext = getRegistryArtifactBackupContext(),
) {
await backupPackageReleaseToObjectStorage(ctx, item, context, {
withIndexWrite: (indexPath, write) =>
withRegistryArtifactBackupIndexLease(ctx, indexPath, write),
});
}
async function withRegistryArtifactBackupIndexLease<T>(
ctx: ActionCtx,
indexPath: string,
run: () => Promise<T>,
) {
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">,