From 7e95dc640bc4557f5b5099dedef16482341374af Mon Sep 17 00:00:00 2001 From: Arunavo Ray Date: Thu, 10 Sep 2026 17:05:03 +0530 Subject: [PATCH] fix(releases): never duplicate release assets on the destination Gitea and Forgejo accept any number of attachments with the same name: CreateReleaseAttachment has no name-collision check, so every upload POST appends another attachment. Asset reconciliation listed what a release already had and then kept uploading for tens of seconds or minutes with no mutual exclusion, so two passes over one repository in the same run both read "missing" and both uploaded. That wide check-to-write window is what left two, three or five copies of every asset (#417). The classifier now groups existing attachments by name, keeps one copy and reports the surplus and stale ones for deletion. Reconciliation fails closed: nothing is uploaded when the attachment list cannot be read, deletes run before any upload, and an asset whose stale copy could not be deleted is skipped until the next sync. Release reconciliation holds a lock keyed on the destination repository, so two passes cannot interleave even when they come from two repository rows. The sync path and both mirror paths claim the row with one conditional UPDATE instead of reading the status and writing it in a second statement. --- CHANGELOG.md | 6 + src/lib/gitea-enhanced.test.ts | 124 ++++++- src/lib/gitea-enhanced.ts | 41 ++- src/lib/gitea-org-mirror-destination.test.ts | 11 +- src/lib/gitea-release-assets.test.ts | 320 +++++++++++++++++ src/lib/gitea-releases.test.ts | 111 +++++- src/lib/gitea.ts | 350 ++++++++++++++----- src/lib/utils/keyed-mutex.test.ts | 115 ++++++ src/lib/utils/keyed-mutex.ts | 63 ++++ src/lib/utils/repo-status-claim.ts | 68 ++++ src/pages/api/job/approve-sync.ts | 60 +++- 11 files changed, 1140 insertions(+), 129 deletions(-) create mode 100644 src/lib/gitea-release-assets.test.ts create mode 100644 src/lib/utils/keyed-mutex.test.ts create mode 100644 src/lib/utils/keyed-mutex.ts create mode 100644 src/lib/utils/repo-status-claim.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index d4bfacf..6ffceea 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - The organization card now shows the default for the configured strategy in the preview, the placeholder, the helper text and the reset button, and marks any stored destination as custom - The repository destination column shows the organization's override as a repository's default, and pinning a repository to the value the strategy already produces is stored rather than discarded - Reset to Default on an organization card clears the override instead of saving the current value again +- Release assets are never duplicated on the destination (#417) + - Gitea and Forgejo accept any number of attachments with the same name, so two mirror passes that overlapped on one repository both saw an asset as missing and both uploaded it, leaving two, three or five copies + - Release reconciliation now runs one pass at a time per destination repository, keeps a single copy of each asset, and deletes surplus and stale copies before uploading anything + - A sync no longer starts on a repository that is already being mirrored or synced, and the mirror path claims the repository in one statement instead of checking and then writing + - Nothing is uploaded when the destination cannot say which assets a release already has, or when a stale copy could not be removed first + - The next sync of an affected repository removes the surplus copies for every release that still gets its assets; releases that have dropped out of the newest-N window keep their duplicates until they are removed by hand - Fixed metadata mirroring authentication errors (#68) - Changed field checking from `username` to `defaultOwner` in metadata functions - Added proper field validation for all metadata operations diff --git a/src/lib/gitea-enhanced.test.ts b/src/lib/gitea-enhanced.test.ts index f0eed1a..b4380dd 100644 --- a/src/lib/gitea-enhanced.test.ts +++ b/src/lib/gitea-enhanced.test.ts @@ -25,6 +25,9 @@ let mockShouldBlockSyncOnBackupFailure = true; // dbUpdateSetCalls so tests can assert on what got written (e.g. archived // repos keeping status "archived" after a Manual Sync). const dbUpdateSetCalls: any[] = []; +// Rows the sync's atomic status claim gets back (#417): one row means this run +// claimed the repository, an empty array means another mirror or sync owns it. +let mockClaimRows: Array<{ id: string }> = [{ id: "repo-1" }]; const mockDb = { insert: mock((table: any) => ({ values: mock((data: any) => Promise.resolve({ insertedId: "mock-id" })) @@ -32,7 +35,13 @@ const mockDb = { update: mock(() => ({ set: mock((data: any) => { dbUpdateSetCalls.push(data); - return { where: mock(() => Promise.resolve()) }; + return { + where: mock(() => { + const result: any = Promise.resolve(); + result.returning = mock(() => Promise.resolve(mockClaimRows)); + return result; + }) + }; }) })) }; @@ -453,6 +462,7 @@ describe("Enhanced Gitea Operations", () => { mockHttpDelete.mockClear(); mockHttpPatch.mockClear(); dbUpdateSetCalls.length = 0; + mockClaimRows = [{ id: "repo-1" }]; mockCreatePreSyncBundleBackup.mockClear(); mockCreatePreSyncBundleBackup.mockImplementation(() => Promise.resolve({ bundlePath: "/tmp/mock.bundle" }) @@ -671,6 +681,118 @@ describe("Enhanced Gitea Operations", () => { expect(mockMirrorGitHubReleasesToGitea).not.toHaveBeenCalled(); }); + test("refuses to start when another run already owns the repository (#417)", async () => { + // Cast: the fixture only sets the fields this path reads. + const config = { + userId: "user123", + githubConfig: { + username: "testuser", + token: "github-token", + privateRepositories: false, + mirrorStarred: true, + }, + giteaConfig: { + url: "https://gitea.example.com", + token: "encrypted-token", + defaultOwner: "testuser", + mirrorReleases: true, + }, + } as unknown as Partial; + + const repository = { + id: "repo456", + name: "mirror-repo", + fullName: "user/mirror-repo", + owner: "user", + cloneUrl: "https://github.com/user/mirror-repo.git", + isPrivate: false, + isStarred: true, + // The row is already "syncing" from another run; the claim loses. + status: repoStatusEnum.parse("syncing"), + visibility: "public", + userId: "user123", + createdAt: new Date(), + updatedAt: new Date(), + } as unknown as Repository; + + // The conditional UPDATE matches no row: another mirror or sync owns it. + mockClaimRows = []; + + const result = await syncGiteaRepoEnhanced( + { config, repository }, + { + getGiteaRepoOwnerAsync: mockGetGiteaRepoOwnerAsync, + mirrorGitHubReleasesToGitea: mockMirrorGitHubReleasesToGitea, + mirrorGitRepoIssuesToGitea: mockMirrorGitRepoIssuesToGitea, + mirrorGitRepoPullRequestsToGitea: mockMirrorGitRepoPullRequestsToGitea, + mirrorGitRepoLabelsToGitea: mockMirrorGitRepoLabelsToGitea, + mirrorGitRepoMilestonesToGitea: mockMirrorGitRepoMilestonesToGitea, + syncRepositoryMetadataToGitea: mockSyncRepositoryMetadataToGitea, + } + ); + + expect(result).toEqual({ skipped: true, reason: "already-in-progress" }); + // Nothing ran: no release or metadata mirroring, and the row was not + // marked failed (it belongs to the run that owns it). + expect(mockMirrorGitHubReleasesToGitea).not.toHaveBeenCalled(); + expect(mockSyncRepositoryMetadataToGitea).not.toHaveBeenCalled(); + expect(dbUpdateSetCalls.some((call) => call.status === "failed")).toBe(false); + }); + + test("syncs when the caller already claimed the row (approve-sync)", async () => { + // Cast: the fixture only sets the fields this path reads. + const config = { + userId: "user123", + githubConfig: { + username: "testuser", + token: "github-token", + privateRepositories: false, + mirrorStarred: true, + }, + giteaConfig: { + url: "https://gitea.example.com", + token: "encrypted-token", + defaultOwner: "testuser", + mirrorReleases: true, + }, + } as unknown as Partial; + + const repository = { + id: "repo456", + name: "mirror-repo", + fullName: "user/mirror-repo", + owner: "user", + cloneUrl: "https://github.com/user/mirror-repo.git", + isPrivate: false, + isStarred: true, + // approve-sync has already moved the row to "syncing" itself. + status: repoStatusEnum.parse("syncing"), + visibility: "public", + userId: "user123", + createdAt: new Date(), + updatedAt: new Date(), + } as unknown as Repository; + + // Would refuse the claim, but alreadyClaimed skips it entirely. + mockClaimRows = []; + + const result = await syncGiteaRepoEnhanced( + { config, repository, alreadyClaimed: true }, + { + getGiteaRepoOwnerAsync: mockGetGiteaRepoOwnerAsync, + mirrorGitHubReleasesToGitea: mockMirrorGitHubReleasesToGitea, + mirrorGitRepoIssuesToGitea: mockMirrorGitRepoIssuesToGitea, + mirrorGitRepoPullRequestsToGitea: mockMirrorGitRepoPullRequestsToGitea, + mirrorGitRepoLabelsToGitea: mockMirrorGitRepoLabelsToGitea, + mirrorGitRepoMilestonesToGitea: mockMirrorGitRepoMilestonesToGitea, + syncRepositoryMetadataToGitea: mockSyncRepositoryMetadataToGitea, + } + ); + + expect(result).toEqual({ success: true }); + expect(mockMirrorGitHubReleasesToGitea).toHaveBeenCalledTimes(1); + }); + test("should successfully sync a mirror repository", async () => { const config: Partial = { userId: "user123", diff --git a/src/lib/gitea-enhanced.ts b/src/lib/gitea-enhanced.ts index b00c852..6535e0b 100644 --- a/src/lib/gitea-enhanced.ts +++ b/src/lib/gitea-enhanced.ts @@ -24,6 +24,7 @@ import { httpPost, httpGet, httpPatch, HttpError } from "./http-client"; import { db, repositories } from "./db"; import { eq } from "drizzle-orm"; import { repoStatusEnum } from "@/types/Repository"; +import { claimRepositoryForInFlightWork } from "./utils/repo-status-claim"; import { resolveMirrorOptionsForRepository } from "./utils/mirror-overrides"; import { createPreSyncBundleBackup, @@ -444,11 +445,19 @@ export async function syncGiteaRepoEnhanced({ config, repository, skipForcePushDetection, + alreadyClaimed, }: { config: Partial; repository: Repository; /** When true, skip force-push detection and blocking (used by approve-sync). */ skipForcePushDetection?: boolean; + /** + * Set by a caller that already claimed this row for the sync with its own + * conditional UPDATE (approve-sync does, so the UI sees "syncing" right + * away). Skips the claim below instead of refusing to run on the caller's + * own "syncing" status. + */ + alreadyClaimed?: boolean; }, deps?: SyncDependencies): Promise { try { if (!config.userId || !config.giteaConfig?.url || !config.giteaConfig?.token) { @@ -472,14 +481,30 @@ export async function syncGiteaRepoEnhanced({ console.log(`[Sync] Starting sync for repository ${repository.name}`); - // Mark repo as "syncing" in DB - await db - .update(repositories) - .set({ - status: repoStatusEnum.parse("syncing"), - updatedAt: new Date(), - }) - .where(eq(repositories.id, repository.id!)); + // Mark repo as "syncing" in DB, but only while no other run owns the row. + // This used to be an unconditional write: a sync would start on a + // repository that was already mirroring or syncing, and the two passes + // reconciled the same releases against the same destination at the same + // time, which duplicated release assets (#417). A row left in an in-flight + // status by a crash is taken over once it goes stale (two hours), and + // resetStuckMirrorStatuses resets those rows to "failed" as well, so + // nothing is locked out permanently. + if (!alreadyClaimed) { + const claimed = await claimRepositoryForInFlightWork({ + repositoryId: repository.id!, + set: { + status: repoStatusEnum.parse("syncing"), + updatedAt: new Date(), + }, + }); + + if (!claimed) { + console.log( + `[Sync] Skipping ${repository.name}: it is already being mirrored or synced by another run.` + ); + return { skipped: true, reason: "already-in-progress" }; + } + } // Resolve sync target in a backward-compatible order: // 1) recorded mirroredLocation (actual historical mirror location) diff --git a/src/lib/gitea-org-mirror-destination.test.ts b/src/lib/gitea-org-mirror-destination.test.ts index 7fc1964..c4e52c3 100644 --- a/src/lib/gitea-org-mirror-destination.test.ts +++ b/src/lib/gitea-org-mirror-destination.test.ts @@ -113,7 +113,16 @@ mock.module("@/lib/db", () => { }), }), update: (_table: any) => ({ - set: (_data: any) => ({ where: (_cond: any) => Promise.resolve() }), + set: (_data: any) => ({ + where: (_cond: any) => { + // The mirror path claims the row with a conditional UPDATE and reads + // the returned rows to find out whether it won (#417); one row means + // claimed. + const result: any = Promise.resolve(); + result.returning = () => Promise.resolve([{ id: "repo-1" }]); + return result; + }, + }), }), insert: (_table: any) => ({ values: (_data: any) => Promise.resolve() }), delete: (_table: any) => ({ where: (_cond: any) => Promise.resolve() }), diff --git a/src/lib/gitea-release-assets.test.ts b/src/lib/gitea-release-assets.test.ts new file mode 100644 index 0000000..6020215 --- /dev/null +++ b/src/lib/gitea-release-assets.test.ts @@ -0,0 +1,320 @@ +/** + * Behavioral tests for release asset reconciliation, regression for #417. + * + * Reported symptom: release assets mirrored to Gitea/Forgejo appear twice (some + * three times, one five times), with identical sizes and identical creation + * times, so the copies came from passes that overlapped inside one run. + * + * Cause: reconcileReleaseAssets listed the attachments a release already had, + * then downloaded and uploaded everything it judged missing. The destination + * has no name-collision check on attachment upload (Gitea's and Forgejo's + * CreateReleaseAttachment appends a row for every POST), so two overlapping + * passes both read "missing" and both uploaded. A failed listing was read as + * "the release has no assets", and a failed delete was followed by the upload + * anyway; both added copies too. + * + * The fake destination below models the real behavior that makes duplicates + * possible: every POST appends another attachment with the same name. + * + * No module mocks here on purpose (bun's mock.module is process-wide and leaks + * into other test files): reconcileReleaseAssets takes the fetch it should use. + */ + +import { describe, expect, it } from "bun:test"; +import { reconcileReleaseAssets, buildReleaseTargetLockKey } from "@/lib/gitea"; +import { withKeyedLock } from "@/lib/utils/keyed-mutex"; +import type { Config } from "@/types/config"; + +const GITEA_URL = "https://gitea.example.com"; +const OWNER = "mirror-owner"; +const REPO = "demo"; +const RELEASE_ID = 77; +const ASSETS_URL = `${GITEA_URL}/api/v1/repos/${OWNER}/${REPO}/releases/${RELEASE_ID}/assets`; + +const config = { + giteaConfig: { url: GITEA_URL, token: "gitea-token", defaultOwner: OWNER }, +} as unknown as Partial; + +const decryptedConfig = { + giteaConfig: { url: GITEA_URL, token: "gitea-token", defaultOwner: OWNER }, + githubConfig: { token: "github-token", username: "upstream" }, +} as unknown as Config; + +type Attachment = { id: number; name: string; size: number }; + +interface FakeDestinationOptions { + /** Attachments the release already carries. */ + existing?: Attachment[]; + /** Status for the "list attachments" GET. */ + listStatus?: number; + /** Status for attachment deletes. */ + deleteStatus?: number; + /** Status for attachment uploads. */ + uploadStatus?: number; +} + +/** A GitHub asset plus the destination-side size the fake records on upload. */ +function githubAsset(name: string, size: number) { + return { + name, + size, + browser_download_url: `https://github.com/upstream/demo/releases/download/v1.0.0/${name}`, + }; +} + +function createFakeDestination(options: FakeDestinationOptions = {}) { + const attachments: Attachment[] = [...(options.existing ?? [])]; + const requests: Array<{ method: string; url: string }> = []; + let nextId = 500; + + const fetchImpl = (async (input: any, init?: any) => { + const url = typeof input === "string" ? input : String(input); + const method = String(init?.method ?? "GET").toUpperCase(); + requests.push({ method, url }); + + // Every request yields the event loop, so a second pass running at the + // same time gets to interleave, which is what created the duplicates. + await new Promise((resolve) => setTimeout(resolve, 0)); + + // Downloading the asset from the source host. + if (url.startsWith("https://github.com/")) { + return new Response(new Uint8Array([1, 2, 3, 4]), { status: 200 }); + } + + if (method === "GET" && url === ASSETS_URL) { + const status = options.listStatus ?? 200; + if (status !== 200) { + return new Response("boom", { status, statusText: "Internal Server Error" }); + } + return new Response(JSON.stringify(attachments), { + status: 200, + headers: { "content-type": "application/json" }, + }); + } + + if (method === "DELETE" && url.startsWith(`${ASSETS_URL}/`)) { + const status = options.deleteStatus ?? 204; + if (status >= 400) { + return new Response("nope", { status, statusText: "Internal Server Error" }); + } + const id = Number(url.slice(`${ASSETS_URL}/`.length)); + const index = attachments.findIndex((a) => a.id === id); + if (index >= 0) attachments.splice(index, 1); + return new Response(null, { status }); + } + + if (method === "POST" && url.startsWith(`${ASSETS_URL}?`)) { + const status = options.uploadStatus ?? 201; + if (status >= 400) { + return new Response("rejected", { status, statusText: "Bad Request" }); + } + const name = decodeURIComponent(new URL(url).searchParams.get("name") ?? ""); + // The real destination does no name-collision check: every POST appends + // another attachment, even when one with that name is already there. + attachments.push({ id: nextId++, name, size: uploadSizes.get(name) ?? 4 }); + return new Response(JSON.stringify({ id: nextId }), { + status, + headers: { "content-type": "application/json" }, + }); + } + + throw new Error(`unexpected request: ${method} ${url}`); + }) as unknown as typeof fetch; + + const uploadSizes = new Map(); + + return { + fetchImpl, + attachments, + requests, + /** Size the destination records for an uploaded asset (defaults to 4). */ + setUploadedSize(name: string, size: number) { + uploadSizes.set(name, size); + }, + posts: () => requests.filter((r) => r.method === "POST"), + deletes: () => requests.filter((r) => r.method === "DELETE"), + }; +} + +function reconcile( + fetchImpl: typeof fetch, + githubAssets: Array<{ name: string; size: number; browser_download_url: string }> +) { + return reconcileReleaseAssets({ + config, + decryptedConfig, + repoOwner: OWNER, + repoName: REPO, + giteaReleaseId: RELEASE_ID, + githubAssets, + tagName: "v1.0.0", + fetchImpl, + }); +} + +describe("reconcileReleaseAssets", () => { + it("uploads every asset once on a release that has none", async () => { + const destination = createFakeDestination(); + destination.setUploadedSize("base.zip", 4); + destination.setUploadedSize("extras.zip", 4); + + const result = await reconcile(destination.fetchImpl, [ + githubAsset("base.zip", 4), + githubAsset("extras.zip", 4), + ]); + + expect(result).toEqual({ uploaded: 2, failed: 0, skipped: 0 }); + expect(destination.posts()).toHaveLength(2); + expect(destination.attachments.map((a) => a.name).sort()).toEqual([ + "base.zip", + "extras.zip", + ]); + }); + + it("two overlapping passes under the destination lock upload each asset once (#417)", async () => { + const destination = createFakeDestination(); + destination.setUploadedSize("base.zip", 4); + const assets = [githubAsset("base.zip", 4)]; + const key = buildReleaseTargetLockKey(GITEA_URL, OWNER, REPO); + + // Two runs of the same repository starting at the same time: the scheduler + // and a manual sync, or two repository rows pointing at one destination. + const [first, second] = await Promise.all([ + withKeyedLock(key, () => reconcile(destination.fetchImpl, assets)), + withKeyedLock(key, () => reconcile(destination.fetchImpl, assets)), + ]); + + expect(destination.posts()).toHaveLength(1); + expect(destination.attachments).toHaveLength(1); + expect(first).toEqual({ uploaded: 1, failed: 0, skipped: 0 }); + // The second pass finds the asset already there and leaves it alone. + expect(second).toEqual({ uploaded: 0, failed: 0, skipped: 1 }); + }); + + it("without the lock, two overlapping passes are what duplicated the asset", async () => { + const destination = createFakeDestination(); + destination.setUploadedSize("base.zip", 4); + const assets = [githubAsset("base.zip", 4)]; + + await Promise.all([ + reconcile(destination.fetchImpl, assets), + reconcile(destination.fetchImpl, assets), + ]); + + // The fake reproduces the reported failure: identical name, identical size, + // two attachments. This is the state the lock prevents. + expect(destination.posts()).toHaveLength(2); + expect(destination.attachments).toHaveLength(2); + }); + + it("uploads nothing when the existing attachments cannot be listed", async () => { + const destination = createFakeDestination({ listStatus: 500 }); + + const result = await reconcile(destination.fetchImpl, [ + githubAsset("base.zip", 4), + githubAsset("extras.zip", 4), + ]); + + // Fail closed: a failed listing used to read as "the release has no + // assets", so the pass re-uploaded everything the release already had. + expect(destination.posts()).toHaveLength(0); + expect(destination.deletes()).toHaveLength(0); + expect(result).toEqual({ uploaded: 0, failed: 2, skipped: 0 }); + }); + + it("does not upload an asset whose stale copy could not be deleted", async () => { + const destination = createFakeDestination({ + existing: [{ id: 11, name: "firmware.bin", size: 1024 }], + deleteStatus: 500, + }); + + const result = await reconcile(destination.fetchImpl, [ + githubAsset("firmware.bin", 2048), + ]); + + expect(destination.deletes()).toHaveLength(1); + // Uploading after the failed delete would leave two copies. + expect(destination.posts()).toHaveLength(0); + expect(destination.attachments).toHaveLength(1); + expect(result).toEqual({ uploaded: 0, failed: 1, skipped: 0 }); + }); + + it("cleans up an existing duplicate without uploading anything", async () => { + const destination = createFakeDestination({ + existing: [ + { id: 11, name: "base.zip", size: 4 }, + { id: 12, name: "base.zip", size: 4 }, + ], + }); + + const result = await reconcile(destination.fetchImpl, [githubAsset("base.zip", 4)]); + + expect(destination.deletes()).toHaveLength(1); + expect(destination.deletes()[0].url).toBe(`${ASSETS_URL}/12`); + expect(destination.posts()).toHaveLength(0); + expect(destination.attachments).toEqual([{ id: 11, name: "base.zip", size: 4 }]); + expect(result).toEqual({ uploaded: 0, failed: 0, skipped: 1 }); + }); + + it("replaces stale copies: deletes both, then uploads once", async () => { + const destination = createFakeDestination({ + existing: [ + { id: 11, name: "firmware.bin", size: 1024 }, + { id: 12, name: "firmware.bin", size: 999 }, + ], + }); + destination.setUploadedSize("firmware.bin", 2048); + + const result = await reconcile(destination.fetchImpl, [ + githubAsset("firmware.bin", 2048), + ]); + + expect(destination.deletes().map((r) => r.url)).toEqual([ + `${ASSETS_URL}/11`, + `${ASSETS_URL}/12`, + ]); + expect(destination.posts()).toHaveLength(1); + expect(destination.attachments).toHaveLength(1); + expect(destination.attachments[0].size).toBe(2048); + expect(result).toEqual({ uploaded: 1, failed: 0, skipped: 0 }); + // Deletes come before the upload, never after it. + const kinds = destination.requests.map((r) => r.method); + expect(kinds.indexOf("POST")).toBeGreaterThan(kinds.lastIndexOf("DELETE")); + }); + + it("leaves attachments alone that the release does not have upstream", async () => { + const destination = createFakeDestination({ + existing: [{ id: 11, name: "from-an-older-sync.zip", size: 4 }], + }); + destination.setUploadedSize("base.zip", 4); + + const result = await reconcile(destination.fetchImpl, [githubAsset("base.zip", 4)]); + + expect(destination.deletes()).toHaveLength(0); + expect(destination.posts()).toHaveLength(1); + expect(result).toEqual({ uploaded: 1, failed: 0, skipped: 0 }); + }); + + it("does nothing at all for a release with no assets", async () => { + const destination = createFakeDestination(); + + const result = await reconcile(destination.fetchImpl, []); + + expect(destination.requests).toHaveLength(0); + expect(result).toEqual({ uploaded: 0, failed: 0, skipped: 0 }); + }); +}); + +describe("buildReleaseTargetLockKey", () => { + it("keys on destination, owner and repository, case-insensitively", () => { + expect(buildReleaseTargetLockKey("https://gitea.example.com/", "Owner", "Repo")).toBe( + buildReleaseTargetLockKey("https://GITEA.example.com", "owner", "repo") + ); + }); + + it("separates different repositories on the same destination", () => { + expect(buildReleaseTargetLockKey(GITEA_URL, OWNER, "one")).not.toBe( + buildReleaseTargetLockKey(GITEA_URL, OWNER, "two") + ); + }); +}); diff --git a/src/lib/gitea-releases.test.ts b/src/lib/gitea-releases.test.ts index 563c3c0..fab2e5f 100644 --- a/src/lib/gitea-releases.test.ts +++ b/src/lib/gitea-releases.test.ts @@ -164,6 +164,11 @@ describe("classifyReleasesForReconciliation", () => { * existing release" and left it at 0. * * Fix: reconcile assets idempotently on both paths via classifyAssetsForReconciliation. + * + * Extended for #417: the destination allows several attachments with the same + * name (no name-collision check in CreateReleaseAttachment), so the classifier + * groups existing attachments by name, keeps one copy and reports the surplus + * ones in toDelete. */ describe("classifyAssetsForReconciliation", () => { it("uploads all assets when the Gitea release has none (the #331 broken state)", () => { @@ -173,13 +178,11 @@ describe("classifyAssetsForReconciliation", () => { ]; const gitea: Array<{ id: number; name: string; size: number }> = []; - const { toUpload, toSkip } = classifyAssetsForReconciliation(github, gitea); + const { toUpload, toSkip, toDelete } = classifyAssetsForReconciliation(github, gitea); expect(toSkip).toEqual([]); - expect(toUpload).toEqual([ - { name: "base.zip", replaceAssetId: null }, - { name: "extras.zip", replaceAssetId: null }, - ]); + expect(toDelete).toEqual([]); + expect(toUpload).toEqual(["base.zip", "extras.zip"]); }); it("backfills only the missing asset when one already exists", () => { @@ -189,10 +192,11 @@ describe("classifyAssetsForReconciliation", () => { ]; const gitea = [{ id: 9, name: "base.zip", size: 40_264_954 }]; - const { toUpload, toSkip } = classifyAssetsForReconciliation(github, gitea); + const { toUpload, toSkip, toDelete } = classifyAssetsForReconciliation(github, gitea); expect(toSkip).toEqual(["base.zip"]); - expect(toUpload).toEqual([{ name: "extras.zip", replaceAssetId: null }]); + expect(toUpload).toEqual(["extras.zip"]); + expect(toDelete).toEqual([]); }); it("is idempotent — skips everything when all assets already match by name+size", () => { @@ -205,28 +209,111 @@ describe("classifyAssetsForReconciliation", () => { { id: 10, name: "extras.zip", size: 37_098_528 }, ]; - const { toUpload, toSkip } = classifyAssetsForReconciliation(github, gitea); + const { toUpload, toSkip, toDelete } = classifyAssetsForReconciliation(github, gitea); expect(toUpload).toEqual([]); expect(toSkip).toEqual(["base.zip", "extras.zip"]); + expect(toDelete).toEqual([]); }); - it("replaces an asset whose size changed upstream (re-upload over the stale copy)", () => { + it("replaces an asset whose size changed upstream (delete the stale copy, upload once)", () => { const github = [{ name: "firmware.bin", size: 2048 }]; const gitea = [{ id: 42, name: "firmware.bin", size: 1024 }]; // truncated/stale - const { toUpload, toSkip } = classifyAssetsForReconciliation(github, gitea); + const { toUpload, toSkip, toDelete } = classifyAssetsForReconciliation(github, gitea); expect(toSkip).toEqual([]); - expect(toUpload).toEqual([{ name: "firmware.bin", replaceAssetId: 42 }]); + expect(toUpload).toEqual(["firmware.bin"]); + expect(toDelete).toEqual([42]); }); it("handles a release with no GitHub assets", () => { - const { toUpload, toSkip } = classifyAssetsForReconciliation( + const { toUpload, toSkip, toDelete } = classifyAssetsForReconciliation( [], [{ id: 1, name: "leftover.zip", size: 10 }] ); expect(toUpload).toEqual([]); expect(toSkip).toEqual([]); + expect(toDelete).toEqual([]); + }); + + // --- #417: duplicate attachments on the destination --- + + it("keeps one copy and deletes the extra when two copies both match (#417)", () => { + const github = [{ name: "base.zip", size: 40_264_954 }]; + const gitea = [ + { id: 9, name: "base.zip", size: 40_264_954 }, + { id: 14, name: "base.zip", size: 40_264_954 }, + ]; + + const { toUpload, toSkip, toDelete } = classifyAssetsForReconciliation(github, gitea); + + expect(toUpload).toEqual([]); + expect(toSkip).toEqual(["base.zip"]); + expect(toDelete).toEqual([14]); + }); + + it("deletes all three copies and uploads once when none matches the size", () => { + const github = [{ name: "firmware.bin", size: 2048 }]; + const gitea = [ + { id: 7, name: "firmware.bin", size: 1024 }, + { id: 8, name: "firmware.bin", size: 1024 }, + { id: 9, name: "firmware.bin", size: 99 }, + ]; + + const { toUpload, toSkip, toDelete } = classifyAssetsForReconciliation(github, gitea); + + expect(toUpload).toEqual(["firmware.bin"]); + expect(toSkip).toEqual([]); + expect(toDelete).toEqual([7, 8, 9]); + }); + + it("keeps the matching copy and deletes the two that do not match", () => { + const github = [{ name: "installer.dmg", size: 5_000 }]; + const gitea = [ + { id: 3, name: "installer.dmg", size: 1_000 }, + { id: 4, name: "installer.dmg", size: 5_000 }, + { id: 5, name: "installer.dmg", size: 2_000 }, + ]; + + const { toUpload, toSkip, toDelete } = classifyAssetsForReconciliation(github, gitea); + + expect(toUpload).toEqual([]); + expect(toSkip).toEqual(["installer.dmg"]); + expect(toDelete).toEqual([3, 5]); + }); + + it("never deletes a destination attachment whose name is not in the GitHub list", () => { + const github = [{ name: "base.zip", size: 10 }]; + const gitea = [ + { id: 1, name: "base.zip", size: 10 }, + { id: 2, name: "from-an-older-release.zip", size: 10 }, + { id: 3, name: "from-an-older-release.zip", size: 99 }, + ]; + + const { toUpload, toSkip, toDelete } = classifyAssetsForReconciliation(github, gitea); + + expect(toUpload).toEqual([]); + expect(toSkip).toEqual(["base.zip"]); + expect(toDelete).toEqual([]); + }); + + it("keeps the lowest id whatever order the destination listed the copies in", () => { + const github = [{ name: "base.zip", size: 10 }]; + const listedOneWay = [ + { id: 31, name: "base.zip", size: 10 }, + { id: 12, name: "base.zip", size: 10 }, + { id: 25, name: "base.zip", size: 10 }, + ]; + const listedTheOtherWay = [...listedOneWay].reverse(); + + const first = classifyAssetsForReconciliation(github, listedOneWay); + const second = classifyAssetsForReconciliation(github, listedTheOtherWay); + + // id 12 survives in both, and the deletes come back in the same order. + expect(first.toDelete).toEqual([25, 31]); + expect(second.toDelete).toEqual([25, 31]); + expect(first.toSkip).toEqual(["base.zip"]); + expect(second.toSkip).toEqual(["base.zip"]); }); }); diff --git a/src/lib/gitea.ts b/src/lib/gitea.ts index 150fbd4..7e13b64 100644 --- a/src/lib/gitea.ts +++ b/src/lib/gitea.ts @@ -40,6 +40,8 @@ import { } from "./utils/mirror-overrides"; import { repositoryDestinationColumns } from "./repo-utils"; import { resolveDestinationIdentity } from "./destination-connection"; +import { withKeyedLock } from "./utils/keyed-mutex"; +import { claimRepositoryForInFlightWork } from "./utils/repo-status-claim"; /** * Helper function to get organization configuration including destination override @@ -875,34 +877,30 @@ export const mirrorGithubRepoToGitea = async ({ console.log(`Mirroring repository ${repository.name}`); - // DOUBLE-CHECK: Final idempotency check right before updating status - // This catches race conditions in the small window between first check and status update - const finalCheck = await isRepoCurrentlyMirroring({ - config, - repoName: targetRepoName, - expectedLocation: targetLocation, - }); - - if (finalCheck) { - console.log( - `[Idempotency] Race condition detected - ${repository.fullName} is now being mirrored by another process. Skipping.` - ); - return; - } - - // Mark repos as "mirroring" in DB + // CLAIM THE ROW: mark it "mirroring" only while no other run owns it. This + // used to be a second isRepoCurrentlyMirroring read followed by an + // unconditional write, so two workers starting in the same tick both read + // "free" and both mirrored the repository (#417). One conditional UPDATE + // closes that window. // CRITICAL: Set mirroredLocation NOW (not after success) so idempotency checks work // This becomes the "target location" - where we intend to mirror to // Without this, the idempotency check can't detect concurrent operations on first mirror - await db - .update(repositories) - .set({ + const claimed = await claimRepositoryForInFlightWork({ + repositoryId: repository.id!, + set: { status: repoStatusEnum.parse("mirroring"), mirroredLocation: targetLocation, ...repositoryDestinationColumns(resolveDestinationIdentity(config)), updatedAt: new Date(), - }) - .where(eq(repositories.id, repository.id!)); + }, + }); + + if (!claimed) { + console.log( + `[Idempotency] Race condition detected - ${repository.fullName} is now being mirrored or synced by another process. Skipping.` + ); + return; + } // Append log for "mirroring" status await createMirrorJob({ @@ -1727,34 +1725,28 @@ export async function mirrorGitHubRepoToGiteaOrg({ // Use clean clone URL without embedded credentials (Forgejo 12+ security requirement) const cloneAddress = repository.cloneUrl; - // DOUBLE-CHECK: Final idempotency check right before updating status - // This catches race conditions in the small window between first check and status update - const finalCheck = await isRepoCurrentlyMirroring({ - config, - repoName: targetRepoName, - expectedLocation: targetLocation, - }); - - if (finalCheck) { - console.log( - `[Idempotency] Race condition detected - ${repository.fullName} is now being mirrored by another process. Skipping.` - ); - return; - } - - // Mark repos as "mirroring" in DB + // CLAIM THE ROW: see the same claim on the user-owner path above. One + // conditional UPDATE replaces the old read-then-write, so two workers + // starting in the same tick cannot both mirror this repository (#417). // CRITICAL: Set mirroredLocation NOW (not after success) so idempotency checks work // This becomes the "target location" - where we intend to mirror to // Without this, the idempotency check can't detect concurrent operations on first mirror - await db - .update(repositories) - .set({ + const claimed = await claimRepositoryForInFlightWork({ + repositoryId: repository.id!, + set: { status: repoStatusEnum.parse("mirroring"), mirroredLocation: targetLocation, ...repositoryDestinationColumns(resolveDestinationIdentity(config)), updatedAt: new Date(), - }) - .where(eq(repositories.id, repository.id!)); + }, + }); + + if (!claimed) { + console.log( + `[Idempotency] Race condition detected - ${repository.fullName} is now being mirrored or synced by another process. Skipping.` + ); + return; + } // Note: "mirroring" status events are handled by the concurrency system // to avoid duplicate events during batch operations @@ -3037,37 +3029,90 @@ export function classifyReleasesForReconciliation( } /** - * Decide which of a GitHub release's assets need (re)uploading to Gitea. + * Decide what to upload, keep and delete for one release's assets. * - * Compared by name: - * - present in Gitea with a matching size -> skip (already mirrored) - * - present with a different size -> upload, replacing the stale copy - * - absent -> upload (fresh) + * Destination attachments are grouped by name because the destination happily + * holds several attachments with the same name: Gitea's and Forgejo's + * CreateReleaseAttachment has no name-collision check, so every POST appends a + * new attachment. Two overlapping mirror passes each saw an asset as missing + * and each uploaded it, which is how releases ended up with two, three or five + * copies of every asset (#417). Grouping means a pass that finds duplicates + * cleans them up instead of adding to them. * - * Pure function so the create/update reconciliation decision is unit-testable - * without hitting the network (regression guard for #331). + * For each GitHub asset, matched by name: + * - one or more copies match the size -> keep the lowest id, skip the + * upload, delete every other copy of that name + * - copies exist but none matches the size -> delete all of them, upload once + * - no copy exists -> upload + * + * A destination attachment whose name is not in the GitHub list is never + * deleted: the asset limit (#311), an older upstream release and manual + * uploads all leave attachments this reconciliation has no claim over. + * + * Pure function so the reconciliation decision is unit-testable without + * hitting the network (regression guard for #331 and #417). */ export function classifyAssetsForReconciliation( githubAssets: Array<{ name: string; size: number }>, giteaAssets: Array<{ id: number; name: string; size: number }> ): { - toUpload: Array<{ name: string; replaceAssetId: number | null }>; + toUpload: string[]; toSkip: string[]; + toDelete: number[]; } { - const existingByName = new Map(giteaAssets.map((a) => [a.name, a])); - const toUpload: Array<{ name: string; replaceAssetId: number | null }> = []; + const existingByName = new Map>(); + for (const existing of giteaAssets) { + const copies = existingByName.get(existing.name); + if (copies) { + copies.push(existing); + } else { + existingByName.set(existing.name, [existing]); + } + } + // Lowest id first, so "keep one copy" and the delete order are deterministic + // whatever order the destination listed the attachments in. + for (const copies of existingByName.values()) { + copies.sort((a, b) => a.id - b.id); + } + + const toUpload: string[] = []; const toSkip: string[] = []; + const toDelete: number[] = []; + const seen = new Set(); for (const asset of githubAssets) { - const existing = existingByName.get(asset.name); - if (existing && existing.size === asset.size) { + if (seen.has(asset.name)) { + continue; + } + seen.add(asset.name); + + const copies = existingByName.get(asset.name) ?? []; + if (copies.length === 0) { + toUpload.push(asset.name); + continue; + } + + const keep = copies.find((copy) => copy.size === asset.size); + if (keep) { + // Already mirrored: keep the first (lowest id) matching copy and drop + // every other attachment carrying that name. toSkip.push(asset.name); + for (const copy of copies) { + if (copy.id !== keep.id) { + toDelete.push(copy.id); + } + } } else { - toUpload.push({ name: asset.name, replaceAssetId: existing ? existing.id : null }); + // Every copy is stale (truncated, or changed upstream): remove them all + // and upload a single fresh one. + for (const copy of copies) { + toDelete.push(copy.id); + } + toUpload.push(asset.name); } } - return { toUpload, toSkip }; + return { toUpload, toSkip, toDelete }; } /** @@ -3080,11 +3125,20 @@ export function classifyAssetsForReconciliation( * failed or were interrupted on first creation therefore stayed permanently * asset-less, and re-syncing could never heal it (#331). * - * Strategy: compare by asset name. Skip assets already present with a matching size; - * (re)upload anything missing, and replace an existing asset whose size differs - * (truncated/changed upstream). Returns per-release counts so the caller can report. + * Never adds a copy of an asset that is already there (#417): + * - the existing attachments are listed first, and if that listing fails the + * whole release is skipped ("fail closed") instead of being uploaded blind; + * - surplus and stale attachments are deleted BEFORE anything is uploaded; + * - an asset whose stale copy could not be deleted is not uploaded either, so + * a failed delete never turns into one more copy. + * + * The caller is expected to hold the per-destination lock (see + * mirrorGitHubReleasesToGitea), which keeps two passes over the same + * destination repository from interleaving their list and upload steps. + * + * Returns per-release counts so the caller can report. */ -async function reconcileReleaseAssets({ +export async function reconcileReleaseAssets({ config, decryptedConfig, repoOwner, @@ -3092,6 +3146,7 @@ async function reconcileReleaseAssets({ giteaReleaseId, githubAssets, tagName, + fetchImpl, }: { config: Partial; decryptedConfig: Config; @@ -3100,6 +3155,8 @@ async function reconcileReleaseAssets({ giteaReleaseId: number; githubAssets: Array<{ name: string; size: number; browser_download_url: string }>; tagName: string; + /** Injected in tests; production uses the global fetch. */ + fetchImpl?: typeof fetch; }): Promise<{ uploaded: number; failed: number; skipped: number }> { let uploaded = 0; let failed = 0; @@ -3109,27 +3166,87 @@ async function reconcileReleaseAssets({ return { uploaded, failed, skipped }; } + const doFetch: typeof fetch = fetchImpl ?? fetch; const giteaBaseUrl = config.giteaConfig!.url; const giteaAuth = { Authorization: `token ${decryptedConfig.giteaConfig!.token}` }; + const assetsUrl = `${giteaBaseUrl}/api/v1/repos/${repoOwner}/${repoName}/releases/${giteaReleaseId}/assets`; - // Fetch existing attachments so we only transfer what's missing or changed. - const existingAssets: Array<{ id: number; name: string; size: number }> = await httpGet( - `${giteaBaseUrl}/api/v1/repos/${repoOwner}/${repoName}/releases/${giteaReleaseId}/assets`, - giteaAuth - ) - .then((r) => (Array.isArray(r?.data) ? r.data : [])) - .catch(() => []); + // FAIL CLOSED: the list of what the release already has is the only thing + // standing between a sync and a second copy of every asset. If it cannot be + // read, upload nothing and let the next sync try again. Treating a failed + // GET as "the release has no assets" is exactly how duplicates appeared. + let existingAssets: Array<{ id: number; name: string; size: number }>; + try { + const listResponse = await doFetch(assetsUrl, { headers: giteaAuth }); + if (!listResponse.ok) { + throw new Error(`HTTP ${listResponse.status} ${listResponse.statusText}`); + } + const body = await listResponse.json(); + if (!Array.isArray(body)) { + throw new Error("the assets endpoint did not return a list"); + } + existingAssets = body; + } catch (listError) { + console.error( + `[Releases] Could not list the existing assets of ${tagName}: ${ + listError instanceof Error ? listError.message : String(listError) + }. Skipping all ${githubAssets.length} asset(s) for this release so no duplicate is created; the next sync retries.` + ); + return { uploaded, failed: githubAssets.length, skipped }; + } - const { toUpload, toSkip } = classifyAssetsForReconciliation( + const { toUpload, toSkip, toDelete } = classifyAssetsForReconciliation( githubAssets, existingAssets ); skipped = toSkip.length; + const nameById = new Map(existingAssets.map((a) => [a.id, a.name])); + // Names that still have a surplus or stale attachment on the destination: + // uploading now would leave one copy too many, so the upload waits. + const undeletedNames = new Set(); + + // Deletes run before any upload, so a release that already has duplicates is + // cleaned up by the very sync that notices them. + for (const assetId of toDelete) { + const name = nameById.get(assetId); + try { + const deleteResponse = await doFetch(`${assetsUrl}/${assetId}`, { + method: "DELETE", + headers: giteaAuth, + }); + // 404 means it is already gone, which is the state we wanted. + if (!deleteResponse.ok && deleteResponse.status !== 404) { + throw new Error(`HTTP ${deleteResponse.status} ${deleteResponse.statusText}`); + } + console.log( + `[Releases] Removed surplus/stale attachment ${name ?? assetId} (id ${assetId}) from ${tagName}` + ); + } catch (deleteError) { + if (name) { + undeletedNames.add(name); + } + console.error( + `[Releases] Failed to delete attachment ${name ?? assetId} (id ${assetId}) from ${tagName}: ${ + deleteError instanceof Error ? deleteError.message : String(deleteError) + }` + ); + } + } + const githubByName = new Map(githubAssets.map((a) => [a.name, a])); - for (const { name, replaceAssetId } of toUpload) { + for (const name of toUpload) { const asset = githubByName.get(name)!; + + if (undeletedNames.has(name)) { + console.warn( + `[Releases] Not uploading ${name} for ${tagName}: its stale copy is still on the destination and uploading would leave two copies. Retrying on the next sync.` + ); + failed++; + continue; + } + try { // Download from GitHub. fetch strips the Authorization header on the // cross-host redirect to GitHub's object storage, so this works for both @@ -3137,7 +3254,7 @@ async function reconcileReleaseAssets({ console.log( `[Releases] Downloading asset: ${asset.name} (${asset.size} bytes) for ${tagName}` ); - const assetResponse = await fetch(asset.browser_download_url, { + const assetResponse = await doFetch(asset.browser_download_url, { headers: { Accept: "application/octet-stream", Authorization: `token ${decryptedConfig.githubConfig!.token}`, @@ -3154,23 +3271,14 @@ async function reconcileReleaseAssets({ const assetData = await assetResponse.arrayBuffer(); - // Gitea rejects a duplicate attachment name, so drop a stale/mismatched - // copy before re-uploading. - if (replaceAssetId !== null) { - await httpDelete( - `${giteaBaseUrl}/api/v1/repos/${repoOwner}/${repoName}/releases/${giteaReleaseId}/assets/${replaceAssetId}`, - giteaAuth - ).catch(() => null); - } - const formData = new FormData(); formData.append("attachment", new Blob([assetData]), asset.name); - const uploadResponse = await fetch( - `${giteaBaseUrl}/api/v1/repos/${repoOwner}/${repoName}/releases/${giteaReleaseId}/assets?name=${encodeURIComponent(asset.name)}`, + const uploadResponse = await doFetch( + `${assetsUrl}?name=${encodeURIComponent(asset.name)}`, { method: "POST", - headers: { Authorization: `token ${decryptedConfig.giteaConfig!.token}` }, + headers: giteaAuth, body: formData, } ); @@ -3198,15 +3306,7 @@ async function reconcileReleaseAssets({ return { uploaded, failed, skipped }; } -export async function mirrorGitHubReleasesToGitea({ - octokit, - repository, - config, - giteaOwner, - giteaRepoName, - releaseLimit: releaseLimitOverride, - releaseAssetLimit: releaseAssetLimitOverride, -}: { +type MirrorReleasesParams = { octokit: Octokit; repository: Repository; config: Partial; @@ -3224,7 +3324,73 @@ export async function mirrorGitHubReleasesToGitea({ * none, `undefined` means resolve it here like the release limit. */ releaseAssetLimit?: number | null; -}) { +}; + +/** + * Lock key for one destination repository's releases: the destination base URL + * plus the owner and repository name it is mirrored as, lowercased. + * + * Keyed on the destination and not on the repository row on purpose. Two rows + * (the same upstream repository imported under two sources) resolve to the same + * destination repository, and a per-row status guard cannot see that (#417). + */ +export function buildReleaseTargetLockKey( + giteaBaseUrl: string, + owner: string, + repoName: string +): string { + return `${giteaBaseUrl.replace(/\/+$/, "")}|${owner}|${repoName}`.toLowerCase(); +} + +/** + * Mirror a repository's releases, and the assets of the newest ones, onto the + * destination. + * + * One pass per destination repository at a time (#417): release reconciliation + * reads what the destination already has and then writes what is missing, and + * the destination appends a new attachment for every upload POST without any + * name-collision check. Two passes that overlap on one repository (the + * scheduler racing a manual sync, recovery resuming a live job, or two + * repository rows pointing at the same destination) both read "missing" and + * both upload, leaving duplicates. The lock makes the read and the write one + * unit per destination repository. + */ +export async function mirrorGitHubReleasesToGitea(params: MirrorReleasesParams) { + const { config, repository, giteaOwner, giteaRepoName } = params; + + if ( + !config.giteaConfig?.defaultOwner || + !config.giteaConfig?.token || + !config.giteaConfig?.url + ) { + throw new Error("Gitea config is incomplete for mirroring releases."); + } + + // Resolved here so the lock key names the destination repository, then passed + // down so the locked body does not resolve it a second time. + const repoOwner = giteaOwner || (await getGiteaRepoOwnerAsync({ config, repository })); + const repoName = giteaRepoName || repository.name; + + return withKeyedLock( + buildReleaseTargetLockKey(config.giteaConfig.url, repoOwner, repoName), + () => + mirrorGitHubReleasesToGiteaLocked({ + ...params, + giteaOwner: repoOwner, + giteaRepoName: repoName, + }) + ); +} + +async function mirrorGitHubReleasesToGiteaLocked({ + octokit, + repository, + config, + giteaOwner, + giteaRepoName, + releaseLimit: releaseLimitOverride, + releaseAssetLimit: releaseAssetLimitOverride, +}: MirrorReleasesParams) { if ( !config.giteaConfig?.defaultOwner || !config.giteaConfig?.token || diff --git a/src/lib/utils/keyed-mutex.test.ts b/src/lib/utils/keyed-mutex.test.ts new file mode 100644 index 0000000..062171e --- /dev/null +++ b/src/lib/utils/keyed-mutex.test.ts @@ -0,0 +1,115 @@ +/** + * Unit tests for the per-key async mutex used to serialize release + * reconciliation per destination repository (issue #417). + */ + +import { describe, expect, it } from "bun:test"; +import { keyedLockCount, withKeyedLock } from "@/lib/utils/keyed-mutex"; + +const tick = (ms = 0) => new Promise((resolve) => setTimeout(resolve, ms)); + +describe("withKeyedLock", () => { + it("runs two callers on the same key strictly one after the other", async () => { + const events: string[] = []; + + const first = withKeyedLock("repo-a", async () => { + events.push("a:start"); + await tick(10); + events.push("a:end"); + }); + const second = withKeyedLock("repo-a", async () => { + events.push("b:start"); + await tick(1); + events.push("b:end"); + }); + + await Promise.all([first, second]); + + expect(events).toEqual(["a:start", "a:end", "b:start", "b:end"]); + }); + + it("runs callers on different keys concurrently", async () => { + const events: string[] = []; + + await Promise.all([ + withKeyedLock("repo-a", async () => { + events.push("a:start"); + await tick(10); + events.push("a:end"); + }), + withKeyedLock("repo-b", async () => { + events.push("b:start"); + await tick(10); + events.push("b:end"); + }), + ]); + + // Both started before either finished, so no serialization across keys. + expect(events.slice(0, 2).sort()).toEqual(["a:start", "b:start"]); + }); + + it("releases the key when the callback throws, and surfaces the error", async () => { + const events: string[] = []; + + const failing = withKeyedLock("repo-c", async () => { + events.push("a:start"); + await tick(1); + throw new Error("upload exploded"); + }); + const following = withKeyedLock("repo-c", async () => { + events.push("b:start"); + return "ran anyway"; + }); + + await expect(failing).rejects.toThrow("upload exploded"); + expect(await following).toBe("ran anyway"); + expect(events).toEqual(["a:start", "b:start"]); + }); + + it("keeps FIFO order for several waiters on one key", async () => { + const order: number[] = []; + + await Promise.all( + [1, 2, 3, 4].map((n) => + withKeyedLock("repo-d", async () => { + order.push(n); + // Later callers ask for a shorter delay: without the lock they would + // finish out of order. + await tick(5 - n); + }) + ) + ); + + expect(order).toEqual([1, 2, 3, 4]); + }); + + it("returns the callback's value and forgets the key once nobody is queued", async () => { + const before = keyedLockCount(); + + const value = await withKeyedLock("repo-e", async () => 42); + expect(value).toBe(42); + + // The cleanup runs in a microtask after the tail settles. + await tick(0); + expect(keyedLockCount()).toBe(before); + }); + + it("serializes a second caller that arrives while the first is running", async () => { + let holderRunning = false; + let sawOverlap = false; + + const holder = withKeyedLock("repo-f", async () => { + holderRunning = true; + await tick(15); + holderRunning = false; + }); + + await tick(1); + const waiter = withKeyedLock("repo-f", async () => { + if (holderRunning) sawOverlap = true; + }); + + await Promise.all([holder, waiter]); + expect(sawOverlap).toBe(false); + }); +}); diff --git a/src/lib/utils/keyed-mutex.ts b/src/lib/utils/keyed-mutex.ts new file mode 100644 index 0000000..a31a3da --- /dev/null +++ b/src/lib/utils/keyed-mutex.ts @@ -0,0 +1,63 @@ +/** + * Per-key async mutex (issue #417). + * + * Release asset reconciliation is a read-then-write against the destination: + * list the attachments a release already has, then upload whatever is missing. + * The destination has no name-collision check on attachment upload (Gitea's and + * Forgejo's CreateReleaseAttachment appends a row for every POST), so two + * passes that overlap on the same release both see an asset as absent and both + * upload it. N overlapping passes leave N copies of every asset. + * + * A per-row status guard cannot close that window on its own: two repository + * rows (the same upstream repository imported under two sources, say) resolve + * to the same destination repository, so the exclusion has to be keyed on the + * destination rather than on the database row. + * + * This lock is process-local, which is where the overlapping passes seen in + * #417 come from: the scheduler racing a manual run, recovery resuming a job + * that is still running, or duplicate repository rows processed together by one + * Promise.all. + */ + +/** + * Tail of the queue for each key. It settles (and never rejects) once the last + * queued caller is done, so the next caller only has to wait on it. + */ +const queues = new Map>(); + +/** + * Run `fn` with exclusive access to `key`. Callers on the same key run one at a + * time in the order they arrived; callers on different keys are unaffected. + * The lock is released whether `fn` resolves or throws, and the caller still + * sees `fn`'s own result or error. + */ +export function withKeyedLock(key: string, fn: () => Promise): Promise { + const previous = queues.get(key) ?? Promise.resolve(); + + // `previous` never rejects, so a caller that threw still hands the key on. + const result = previous.then(() => fn()); + + // The tail swallows the outcome: the next caller only needs to know the + // previous one finished, and an unobserved rejection here would surface as + // an unhandled rejection. + const tail = result.then( + () => undefined, + () => undefined + ); + queues.set(key, tail); + + // Drop the key once nobody is queued behind this caller, so the map does not + // grow one permanent entry per repository. + void tail.then(() => { + if (queues.get(key) === tail) { + queues.delete(key); + } + }); + + return result; +} + +/** Number of keys currently held or queued. For tests and diagnostics. */ +export function keyedLockCount(): number { + return queues.size; +} diff --git a/src/lib/utils/repo-status-claim.ts b/src/lib/utils/repo-status-claim.ts new file mode 100644 index 0000000..6cf0803 --- /dev/null +++ b/src/lib/utils/repo-status-claim.ts @@ -0,0 +1,68 @@ +/** + * Atomic claim of a repository row for an in-flight operation (issue #417). + * + * Mirroring and syncing both used to read the row's status, decide the + * repository was free, and then write the in-flight status in a second + * statement. Two workers that started in the same tick both read "free" and + * both went on to mirror the same repository, which is one of the ways release + * assets ended up duplicated on the destination. + * + * The claim below is a single conditional UPDATE: the row moves into the + * in-flight status only if it is not already "mirroring" or "syncing", and the + * caller learns from the returned rows whether it won. A row whose in-flight + * status has gone stale (older than the same two-hour window + * isRepoCurrentlyMirroring has always used) can be taken over, so a process + * that died mid-operation never locks a repository out permanently. Rows like + * that are also reset to "failed" by resetStuckMirrorStatuses + * (src/lib/stuck-status-recovery.ts), which runs on startup, in recovery and in + * the scheduler loop. + */ + +import { and, eq, lt, notInArray, or, type SQL } from "drizzle-orm"; +import { db, repositories } from "@/lib/db"; +import { + IN_FLIGHT_REPO_STATUSES, + STUCK_IN_FLIGHT_THRESHOLD_MS, +} from "@/lib/stuck-status-recovery"; + +/** + * WHERE clause that matches one repository row only while no other run owns it. + */ +export function repositoryClaimCondition( + repositoryId: string, + now: Date = new Date() +): SQL | undefined { + const staleCutoff = new Date(now.getTime() - STUCK_IN_FLIGHT_THRESHOLD_MS); + + return and( + eq(repositories.id, repositoryId), + or( + notInArray(repositories.status, [...IN_FLIGHT_REPO_STATUSES]), + lt(repositories.updatedAt, staleCutoff) + ) + ); +} + +/** + * Move one repository row into an in-flight status, but only if no other run + * already owns it. Returns true when this caller claimed the row and false when + * another mirror or sync is in flight for it. + */ +export async function claimRepositoryForInFlightWork({ + repositoryId, + set, + now, +}: { + repositoryId: string; + /** The full update payload, including the in-flight status itself. */ + set: Partial; + now?: Date; +}): Promise { + const claimed = await db + .update(repositories) + .set(set) + .where(repositoryClaimCondition(repositoryId, now)) + .returning({ id: repositories.id }); + + return claimed.length > 0; +} diff --git a/src/pages/api/job/approve-sync.ts b/src/pages/api/job/approve-sync.ts index 0f9b945..ad0b1a5 100644 --- a/src/pages/api/job/approve-sync.ts +++ b/src/pages/api/job/approve-sync.ts @@ -113,9 +113,49 @@ export const POST: APIRoute = async ({ request, locals }) => { // action === "approve": create backup first (safety), then trigger sync const decryptedConfig = decryptConfigTokens(config as unknown as Config); + // Claim the rows before the background work starts: one conditional UPDATE + // per repository, so the status flips to "syncing" immediately (the UI asks + // for that) without stealing a row that a mirror or sync already owns + // (#417). The sync itself is then told the row is claimed. + const claimedRepos: typeof repos = []; + for (const repo of repos) { + const claimed = await db + .update(repositories) + .set({ + status: "syncing", + errorMessage: null, + updatedAt: new Date(), + }) + .where( + and( + eq(repositories.id, repo.id), + eq(repositories.status, "pending-approval"), + ), + ) + .returning({ id: repositories.id }); + + if (claimed.length > 0) { + claimedRepos.push(repo); + } else { + console.log( + `[ApproveSync] Skipping ${repo.name}: it is no longer waiting for approval (another run may have picked it up).`, + ); + } + } + + if (claimedRepos.length === 0) { + return new Response( + JSON.stringify({ + success: false, + message: "No pending-approval repositories were available to sync.", + }), + { status: 409, headers: { "Content-Type": "application/json" } }, + ); + } + // Process in background setTimeout(async () => { - for (const repo of repos) { + for (const repo of claimedRepos) { try { const { getGiteaRepoOwnerAsync } = await import("@/lib/gitea"); const repoOwner = await getGiteaRepoOwnerAsync({ config, repository: repo }); @@ -163,6 +203,8 @@ export const POST: APIRoute = async ({ request, locals }) => { config, repository: repoData, skipForcePushDetection: true, + // The row was claimed above, before this background task started. + alreadyClaimed: true, }); console.log(`[ApproveSync] Sync completed for approved repository: ${repo.name}`); } catch (error) { @@ -174,23 +216,11 @@ export const POST: APIRoute = async ({ request, locals }) => { } }, 0); - // Immediately update status to syncing for responsiveness - for (const repo of repos) { - await db - .update(repositories) - .set({ - status: "syncing", - errorMessage: null, - updatedAt: new Date(), - }) - .where(eq(repositories.id, repo.id)); - } - return new Response( JSON.stringify({ success: true, - message: `Approved sync for ${repos.length} repository(ies). Backup + sync started.`, - repositories: repos.map((repo) => ({ + message: `Approved sync for ${claimedRepos.length} repository(ies). Backup + sync started.`, + repositories: claimedRepos.map((repo) => ({ ...repo, status: "syncing", errorMessage: null,