mirror of
https://github.com/RayLabsHQ/gitea-mirror.git
synced 2026-11-05 16:52:22 +01:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Arunavo Ray
|
7e95dc640b |
No files matched your search
@@ -42,15 +42,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
- Comprehensive fix report documentation
|
||||
|
||||
### Fixed
|
||||
- Crash recovery no longer resumes a job that is still running
|
||||
- A job older than two hours was treated as interrupted even while it checkpointed every two minutes, so recovery started a second pass over the same repositories alongside the original; the age rule is gone and only a missing or stale checkpoint marks a job interrupted
|
||||
- Every completed item is now recorded in the job's checkpoint instead of one in every N, so a real resume skips exactly what was done; progress events are still throttled
|
||||
- Concurrent items completing at the same time could overwrite each other's checkpoint; job progress writes are serialized per job
|
||||
- A resumed job records a fresh start time
|
||||
- The same upstream repository is no longer imported twice under two sources that point at the same host
|
||||
- Discovery decided what was new by (source, full name), so an organization pinned to a public-only github.com source and a personal github.com token source each inserted their own row for one repository, and both rows mirrored to the same destination in the same scheduler batch
|
||||
- Auto import, organization re-discovery and the two import endpoints now identify a repository by its host and full name whichever source lists it; the same name on a different host is still a separate repository
|
||||
- A scheduler pass mirrors or syncs one row per upstream repository and logs the duplicate rows it skips, so existing duplicates never run side by side
|
||||
- A per-organization Mirror Destination equal to the organization's own name is kept instead of being dropped (#416)
|
||||
- The editor treated a typed value matching the organization name as "reset to default" and saved no override, which only holds under the preserve strategy; under single-org the default is the destination organization, so the repositories went there
|
||||
- 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
|
||||
|
||||
@@ -6,7 +6,7 @@
|
||||
"name": "gitea-mirror",
|
||||
"dependencies": {
|
||||
"@astrojs/mdx": "^7.0.3",
|
||||
"@astrojs/node": "^11.1.5",
|
||||
"@astrojs/node": "^11.0.2",
|
||||
"@astrojs/react": "^6.0.1",
|
||||
"@better-auth/api-key": "1.7.2",
|
||||
"@better-auth/oauth-provider": "1.7.2",
|
||||
@@ -94,7 +94,6 @@
|
||||
"lodash": "^4.18.1",
|
||||
"picomatch": "^4.0.4",
|
||||
"sharp": "^0.35.4",
|
||||
"smol-toml": "^1.7.1",
|
||||
"svgo": "^4.1.0",
|
||||
},
|
||||
"packages": {
|
||||
@@ -144,7 +143,7 @@
|
||||
|
||||
"@astrojs/mdx": ["@astrojs/mdx@7.0.3", "", { "dependencies": { "@astrojs/internal-helpers": "0.10.1", "@astrojs/markdown-remark": "7.2.1", "@mdx-js/mdx": "^3.1.1", "acorn": "^8.16.0", "es-module-lexer": "^2.0.0", "estree-util-visit": "^2.0.0", "hast-util-to-html": "^9.0.5", "piccolore": "^0.1.3", "rehype-raw": "^7.0.0", "remark-gfm": "^4.0.1", "remark-smartypants": "^3.0.2", "source-map": "^0.7.6", "unist-util-visit": "^5.1.0", "vfile": "^6.0.3" }, "peerDependencies": { "@astrojs/markdown-satteri": "^0.3.1", "astro": "^7.0.0" }, "optionalPeers": ["@astrojs/markdown-satteri"] }, "sha512-RxyIwU0uFam5ftwqKOjpIdhnFxZ/kEikeimLyQy3eGXbHT8WgRGzzesOIHVU8+m9TY8ag5WVOyvV24/GyqPdPQ=="],
|
||||
|
||||
"@astrojs/node": ["@astrojs/node@11.1.5", "", { "dependencies": { "@astrojs/internal-helpers": "0.11.0", "send": "^1.2.1", "server-destroy": "^1.0.1" }, "peerDependencies": { "astro": "^7.2.1" } }, "sha512-CgA+LmG4UWqHxg/Vdc5MdxjjJNZSjKYQQx4uEUwnbj+1ZD2CYumJI3OWg2twsVdj2FlYsB5yvmCC54Re+OK+UQ=="],
|
||||
"@astrojs/node": ["@astrojs/node@11.0.2", "", { "dependencies": { "@astrojs/internal-helpers": "0.10.1", "send": "^1.2.1", "server-destroy": "^1.0.1" }, "peerDependencies": { "astro": "^7.0.0" } }, "sha512-/ijULxT+A5Cm8wSwWZ2vgqfim1b05D6B8n/a9l6MMA4FCotIH73g7fL7y76XojKXpTe75FVvQH92OxsMqea9kQ=="],
|
||||
|
||||
"@astrojs/prism": ["@astrojs/prism@4.0.2", "", { "dependencies": { "prismjs": "^1.30.0" } }, "sha512-KTivpmnz6lDsC6o9H4+DNm2SrE/GHzw8cNAvEJwAvUT+eoaEnn/4NtbDNfRRaxaJHdp15gf+tfHAWiXR4wB3BA=="],
|
||||
|
||||
@@ -1474,7 +1473,7 @@
|
||||
|
||||
"sisteransi": ["sisteransi@1.0.5", "", {}, "sha512-bLGGlR1QxBcynn2d5YmDX4MGjlZvy2MRBDRNHLJ8VI6l6+9FUiyTFNJ0IveOSP0bcXgVDPRcfGqA0pjaqUpfVg=="],
|
||||
|
||||
"smol-toml": ["smol-toml@1.8.0", "", {}, "sha512-kCZr2V3ch9i00x8zXRhjUNVcjG9ijES5dDudkXvUVCT5QlJNQWElSJdZqyPemffHoLNUYwOcou0Fy+ojN0uHSQ=="],
|
||||
"smol-toml": ["smol-toml@1.6.0", "", {}, "sha512-4zemZi0HvTnYwLfrpk/CF9LOd9Lt87kAt50GnqhMpyF9U3poDAP2+iukq2bZsO/ufegbYehBkqINbsWxj4l4cw=="],
|
||||
|
||||
"sonner": ["sonner@2.0.7", "", { "peerDependencies": { "react": "^18.0.0 || ^19.0.0 || ^19.0.0-rc", "react-dom": "^18.0.0 || ^19.0.0 || ^19.0.0-rc" } }, "sha512-W6ZN4p58k8aDKA4XPcx2hpIQXBRAgyiWVkYhT7CvK6D3iAu7xjvVyhQHg2/iaKJZ1XVJ4r7XuwGL+WGEK37i9w=="],
|
||||
|
||||
@@ -1696,8 +1695,6 @@
|
||||
|
||||
"@astrojs/compiler-binding-wasm32-wasi/@napi-rs/wasm-runtime": ["@napi-rs/wasm-runtime@1.2.3", "", { "dependencies": { "@tybys/wasm-util": "^0.10.3" }, "peerDependencies": { "@emnapi/core": "^1.7.1 || ^2.0.0-alpha.4", "@emnapi/runtime": "^1.7.1 || ^2.0.0-alpha.4" } }, "sha512-UMduMbqO5s5zF2NkNacMT/yK5Y5QiKvWr2+50bzIIxFDwVJ2h49b+oyjaCGPhJxd2/gC2x39EHv/gHVuu36x2Q=="],
|
||||
|
||||
"@astrojs/node/@astrojs/internal-helpers": ["@astrojs/internal-helpers@0.11.0", "", { "dependencies": { "@types/hast": "^3.0.4", "@types/mdast": "^4.0.4", "js-yaml": "^4.3.0", "picomatch": "^4.0.4", "retext-smartypants": "^6.2.0", "shiki": "^4.0.2", "smol-toml": "^1.6.0", "unified": "^11.0.5" } }, "sha512-3rzxJ+xbo0+8YyqOzLziIN32wmsHdCjEVz2sGOpRxJ+Ben/KiLph4ItxBy1abEL+E8fkRzqjg0rfXmaHJGw9JA=="],
|
||||
|
||||
"@astrojs/react/@vitejs/plugin-react": ["@vitejs/plugin-react@5.2.0", "", { "dependencies": { "@babel/core": "^7.29.0", "@babel/plugin-transform-react-jsx-self": "^7.27.1", "@babel/plugin-transform-react-jsx-source": "^7.27.1", "@rolldown/pluginutils": "1.0.0-rc.3", "@types/babel__core": "^7.20.5", "react-refresh": "^0.18.0" }, "peerDependencies": { "vite": "^4.2.0 || ^5.0.0 || ^6.0.0 || ^7.0.0 || ^8.0.0" } }, "sha512-YmKkfhOAi3wsB1PhJq5Scj3GXMn3WvtQ/JC0xoopuHoXSdmtdStOpFrYaT1kie2YgFBcIe64ROzMYRjCrYOdYw=="],
|
||||
|
||||
"@astrojs/react/vite": ["vite@8.1.5", "", { "dependencies": { "lightningcss": "^1.32.0", "picomatch": "^4.0.5", "postcss": "^8.5.17", "rolldown": "~1.1.5", "tinyglobby": "^0.2.17" }, "optionalDependencies": { "fsevents": "~2.3.3" }, "peerDependencies": { "@types/node": "^20.19.0 || >=22.12.0", "@vitejs/devtools": "^0.3.0", "esbuild": "^0.27.0 || ^0.28.0", "jiti": ">=1.21.0", "less": "^4.0.0", "sass": "^1.70.0", "sass-embedded": "^1.70.0", "stylus": ">=0.54.8", "sugarss": "^5.0.0", "terser": "^5.16.0", "tsx": "^4.8.1", "yaml": "^2.4.2" }, "optionalPeers": ["@types/node", "@vitejs/devtools", "esbuild", "jiti", "less", "sass", "sass-embedded", "stylus", "sugarss", "terser", "tsx", "yaml"], "bin": { "vite": "bin/vite.js" } }, "sha512-7ULLwsCdYx/nRyrpiEwvqb5TFHrMVZyBt+rg/OAXT7rgj/z+DtTDyKFeLAdDkubDVDKD8jOsndmy7m55XcfUsw=="],
|
||||
|
||||
@@ -113,9 +113,9 @@
|
||||
url = "https://registry.npmjs.org/@astrojs/mdx/-/mdx-7.0.3.tgz";
|
||||
hash = "sha512-RxyIwU0uFam5ftwqKOjpIdhnFxZ/kEikeimLyQy3eGXbHT8WgRGzzesOIHVU8+m9TY8ag5WVOyvV24/GyqPdPQ==";
|
||||
};
|
||||
"@astrojs/node@11.1.5" = fetchurl {
|
||||
url = "https://registry.npmjs.org/@astrojs/node/-/node-11.1.5.tgz";
|
||||
hash = "sha512-CgA+LmG4UWqHxg/Vdc5MdxjjJNZSjKYQQx4uEUwnbj+1ZD2CYumJI3OWg2twsVdj2FlYsB5yvmCC54Re+OK+UQ==";
|
||||
"@astrojs/node@11.0.2" = fetchurl {
|
||||
url = "https://registry.npmjs.org/@astrojs/node/-/node-11.0.2.tgz";
|
||||
hash = "sha512-/ijULxT+A5Cm8wSwWZ2vgqfim1b05D6B8n/a9l6MMA4FCotIH73g7fL7y76XojKXpTe75FVvQH92OxsMqea9kQ==";
|
||||
};
|
||||
"@astrojs/prism@4.0.2" = fetchurl {
|
||||
url = "https://registry.npmjs.org/@astrojs/prism/-/prism-4.0.2.tgz";
|
||||
@@ -3205,9 +3205,9 @@
|
||||
url = "https://registry.npmjs.org/sisteransi/-/sisteransi-1.0.5.tgz";
|
||||
hash = "sha512-bLGGlR1QxBcynn2d5YmDX4MGjlZvy2MRBDRNHLJ8VI6l6+9FUiyTFNJ0IveOSP0bcXgVDPRcfGqA0pjaqUpfVg==";
|
||||
};
|
||||
"smol-toml@1.8.0" = fetchurl {
|
||||
url = "https://registry.npmjs.org/smol-toml/-/smol-toml-1.8.0.tgz";
|
||||
hash = "sha512-kCZr2V3ch9i00x8zXRhjUNVcjG9ijES5dDudkXvUVCT5QlJNQWElSJdZqyPemffHoLNUYwOcou0Fy+ojN0uHSQ==";
|
||||
"smol-toml@1.6.0" = fetchurl {
|
||||
url = "https://registry.npmjs.org/smol-toml/-/smol-toml-1.6.0.tgz";
|
||||
hash = "sha512-4zemZi0HvTnYwLfrpk/CF9LOd9Lt87kAt50GnqhMpyF9U3poDAP2+iukq2bZsO/ufegbYehBkqINbsWxj4l4cw==";
|
||||
};
|
||||
"sonner@2.0.7" = fetchurl {
|
||||
url = "https://registry.npmjs.org/sonner/-/sonner-2.0.7.tgz";
|
||||
|
||||
@@ -2,8 +2,8 @@ apiVersion: v2
|
||||
name: gitea-mirror
|
||||
description: Kubernetes helm chart for gitea-mirror
|
||||
type: application
|
||||
version: 3.36.1
|
||||
appVersion: "3.36.1"
|
||||
version: 3.35.1
|
||||
appVersion: "3.35.1"
|
||||
icon: https://github.com/RayLabsHQ/gitea-mirror/blob/main/.github/assets/logo.png
|
||||
keywords:
|
||||
- git
|
||||
|
||||
+2
-3
@@ -1,7 +1,7 @@
|
||||
{
|
||||
"name": "gitea-mirror",
|
||||
"type": "module",
|
||||
"version": "3.36.1",
|
||||
"version": "3.35.1",
|
||||
"engines": {
|
||||
"bun": ">=1.2.9"
|
||||
},
|
||||
@@ -51,7 +51,6 @@
|
||||
"fast-uri": "^3.1.5",
|
||||
"fast-xml-builder": "^1.1.7",
|
||||
"js-yaml": "^4.3.2",
|
||||
"smol-toml": "^1.7.1",
|
||||
"kysely": "^0.28.17",
|
||||
"lodash": "^4.18.1",
|
||||
"picomatch": "^4.0.4",
|
||||
@@ -60,7 +59,7 @@
|
||||
},
|
||||
"dependencies": {
|
||||
"@astrojs/mdx": "^7.0.3",
|
||||
"@astrojs/node": "^11.1.5",
|
||||
"@astrojs/node": "^11.0.2",
|
||||
"@astrojs/react": "^6.0.1",
|
||||
"@better-auth/api-key": "1.7.2",
|
||||
"@better-auth/oauth-provider": "1.7.2",
|
||||
|
||||
+12
-43
@@ -4,7 +4,6 @@ import { eq, and } from "drizzle-orm";
|
||||
import { buildInterruptedJobsCondition } from "./interrupted-job-detection";
|
||||
import { v4 as uuidv4 } from "uuid";
|
||||
import { publishEvent } from "./events";
|
||||
import { withKeyedLock } from "./utils/keyed-mutex";
|
||||
import { triggerJobNotification } from "./notification-service";
|
||||
|
||||
export async function createMirrorJob({
|
||||
@@ -114,34 +113,9 @@ export async function createMirrorJob({
|
||||
}
|
||||
|
||||
/**
|
||||
* Updates the progress of a mirror job.
|
||||
*
|
||||
* Progress is a read-modify-write of the completedItemIds list, and items
|
||||
* finish concurrently, so the update runs under a per-job lock: two items
|
||||
* completing in the same tick used to read the same list and one of them
|
||||
* silently overwrote the other. A resume then re-processed the lost item.
|
||||
*
|
||||
* `silent` records the progress without publishing an event. Every completed
|
||||
* item is checkpointed so a resume knows exactly what is done; the events the
|
||||
* dashboard listens to are throttled by the caller instead.
|
||||
* Updates the progress of a mirror job
|
||||
*/
|
||||
export async function updateMirrorJobProgress(params: {
|
||||
jobId: string;
|
||||
completedItemId?: string;
|
||||
status?: RepoStatus;
|
||||
message?: string;
|
||||
details?: string;
|
||||
inProgress?: boolean;
|
||||
isCompleted?: boolean;
|
||||
/** Record a new start time, as when recovery resumes the job. */
|
||||
startedAt?: Date;
|
||||
/** Write the progress without publishing a mirror-status event. */
|
||||
silent?: boolean;
|
||||
}) {
|
||||
return withKeyedLock(`mirror-job:${params.jobId}`, () => updateMirrorJobProgressLocked(params));
|
||||
}
|
||||
|
||||
async function updateMirrorJobProgressLocked({
|
||||
export async function updateMirrorJobProgress({
|
||||
jobId,
|
||||
completedItemId,
|
||||
status,
|
||||
@@ -149,9 +123,15 @@ async function updateMirrorJobProgressLocked({
|
||||
details,
|
||||
inProgress,
|
||||
isCompleted,
|
||||
startedAt,
|
||||
silent = false,
|
||||
}: Parameters<typeof updateMirrorJobProgress>[0]) {
|
||||
}: {
|
||||
jobId: string;
|
||||
completedItemId?: string;
|
||||
status?: RepoStatus;
|
||||
message?: string;
|
||||
details?: string;
|
||||
inProgress?: boolean;
|
||||
isCompleted?: boolean;
|
||||
}) {
|
||||
try {
|
||||
// Get the current job
|
||||
const [job] = await db
|
||||
@@ -177,10 +157,6 @@ async function updateMirrorJobProgressLocked({
|
||||
}
|
||||
}
|
||||
|
||||
if (startedAt) {
|
||||
updates.startedAt = startedAt;
|
||||
}
|
||||
|
||||
// Update status if provided
|
||||
if (status) {
|
||||
updates.status = status;
|
||||
@@ -219,10 +195,6 @@ async function updateMirrorJobProgressLocked({
|
||||
...updates,
|
||||
};
|
||||
|
||||
if (silent) {
|
||||
return updatedJob;
|
||||
}
|
||||
|
||||
// Create deduplication key for progress updates
|
||||
let deduplicationKey: string | undefined;
|
||||
if (completedItemId) {
|
||||
@@ -352,15 +324,12 @@ export async function resumeInterruptedJob(job: any) {
|
||||
return null;
|
||||
}
|
||||
|
||||
// Update the job to show it's being resumed. The start time moves to
|
||||
// now so the row records this pass; the fresh checkpoint written here is
|
||||
// what keeps a second recovery sweep from resuming the same job again.
|
||||
// Update the job to show it's being resumed
|
||||
await updateMirrorJobProgress({
|
||||
jobId: job.id,
|
||||
message: `Resuming job with ${remainingItemIds.length} remaining items`,
|
||||
details: `Job was interrupted and is being resumed. ${job.completedItemIds.length} of ${job.itemIds.length} items were already processed.`,
|
||||
inProgress: true,
|
||||
startedAt: new Date(),
|
||||
});
|
||||
|
||||
return {
|
||||
|
||||
@@ -18,6 +18,7 @@ import { drizzle } from "drizzle-orm/bun-sqlite";
|
||||
import { mirrorJobs } from "@/lib/db/schema";
|
||||
import {
|
||||
INTERRUPTED_CHECKPOINT_AGE_MS,
|
||||
STALE_JOB_AGE_MS,
|
||||
JOB_HEARTBEAT_INTERVAL_MS,
|
||||
computeInterruptedJobCutoffs,
|
||||
isJobInterrupted,
|
||||
@@ -31,15 +32,14 @@ const NOW = new Date("2026-09-02T12:00:00Z");
|
||||
const ago = (ms: number) => new Date(NOW.getTime() - ms);
|
||||
|
||||
describe("computeInterruptedJobCutoffs", () => {
|
||||
test("keeps the 10 minute checkpoint window and has no stale age cutoff", () => {
|
||||
test("keeps the 10 minute checkpoint window and 2 hour stale window", () => {
|
||||
expect(INTERRUPTED_CHECKPOINT_AGE_MS).toBe(10 * MINUTE);
|
||||
expect(STALE_JOB_AGE_MS).toBe(2 * HOUR);
|
||||
|
||||
const cutoffs = computeInterruptedJobCutoffs(NOW);
|
||||
|
||||
expect(cutoffs.checkpointCutoff.getTime()).toBe(NOW.getTime() - 10 * MINUTE);
|
||||
// The two hour "stale regardless of checkpoint" rule resumed live jobs
|
||||
// and is gone; a checkpointing job is alive however old it is.
|
||||
expect(Object.keys(cutoffs)).toEqual(["checkpointCutoff"]);
|
||||
expect(cutoffs.staleCutoff.getTime()).toBe(NOW.getTime() - 2 * HOUR);
|
||||
});
|
||||
|
||||
test("heartbeat interval leaves room for a missed beat inside the checkpoint window", () => {
|
||||
@@ -78,18 +78,9 @@ describe("isJobInterrupted", () => {
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
test("a job that keeps checkpointing stays live however long ago it started", () => {
|
||||
test("a job running past the stale window is interrupted even with a fresh checkpoint", () => {
|
||||
expect(
|
||||
isJobInterrupted({ inProgress: true, startedAt: ago(3 * HOUR), lastCheckpoint: ago(MINUTE) }, NOW)
|
||||
).toBe(false);
|
||||
expect(
|
||||
isJobInterrupted({ inProgress: true, startedAt: ago(30 * HOUR), lastCheckpoint: ago(2 * MINUTE) }, NOW)
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
test("an old job whose checkpoint went stale is still interrupted", () => {
|
||||
expect(
|
||||
isJobInterrupted({ inProgress: true, startedAt: ago(3 * HOUR), lastCheckpoint: ago(11 * MINUTE) }, NOW)
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
@@ -244,22 +235,6 @@ describe("wiring", () => {
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
test("resumeInterruptedJob records a fresh start time for the resumed pass", () => {
|
||||
const resumeBody = HELPERS_SRC.slice(HELPERS_SRC.indexOf("export async function resumeInterruptedJob("));
|
||||
expect(/inProgress:\s*true,\s*startedAt:\s*new Date\(\)/.test(resumeBody)).toBe(true);
|
||||
});
|
||||
|
||||
test("updateMirrorJobProgress serializes writes per job so concurrent items cannot drop each other", () => {
|
||||
expect(
|
||||
/withKeyedLock\(`mirror-job:\$\{params\.jobId\}`,\s*\(\)\s*=>\s*updateMirrorJobProgressLocked\(params\)\)/.test(HELPERS_SRC)
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
test("processWithRetry checkpoints every completed item instead of the one that trips the interval", () => {
|
||||
expect(/await onCheckpoint\(jobId,\s*itemId,\s*publish\)/.test(CONCURRENCY_SRC)).toBe(true);
|
||||
expect(CONCURRENCY_SRC.includes("onCheckpoint(jobId, 'final')")).toBe(false);
|
||||
});
|
||||
|
||||
test("middleware keeps the recovery latch until the recovery promise settles", () => {
|
||||
expect(
|
||||
/recoveryPromise\s*\.catch\(\(\)\s*=>\s*false\)\s*\.finally\(\(\)\s*=>\s*\{\s*recoveryInFlight\s*=\s*false;\s*\}\)/.test(
|
||||
|
||||
@@ -2,20 +2,11 @@
|
||||
* Decides which in-progress mirror jobs count as interrupted and should be
|
||||
* resumed by recovery (issue #372).
|
||||
*
|
||||
* A job is interrupted when either of these holds:
|
||||
* A job is interrupted when any of these hold:
|
||||
* - it has a checkpoint older than the checkpoint window, or
|
||||
* - it has no checkpoint and is older than the checkpoint window (or has
|
||||
* no recorded start at all, as legacy rows may).
|
||||
*
|
||||
* A job that keeps checkpointing is alive, however long ago it started.
|
||||
* Running jobs refresh their checkpoint on a heartbeat between items, so a
|
||||
* missing or old checkpoint is the only evidence that the process behind a
|
||||
* job is gone. There used to be a third rule, "started more than two hours
|
||||
* ago, whatever the checkpoint says", which resumed large jobs while they
|
||||
* were still running: the resumed pass then re-processed the same
|
||||
* repositories alongside the original, which is one way release assets got
|
||||
* duplicated (#417). A job that hangs without ever finishing an item is a
|
||||
* different problem and is not what recovery is for.
|
||||
* no recorded start at all, as legacy rows may), or
|
||||
* - it started longer ago than the stale window, whatever its checkpoint.
|
||||
*
|
||||
* The grace period for jobs without a checkpoint is the point of this
|
||||
* module. Checkpoints are written when an item completes, so a job that has
|
||||
@@ -33,6 +24,9 @@ import type { mirrorJobs } from "@/lib/db/schema";
|
||||
/** A job with no checkpoint newer than this is considered inactive. */
|
||||
export const INTERRUPTED_CHECKPOINT_AGE_MS = 10 * 60 * 1000;
|
||||
|
||||
/** A job that started longer ago than this is considered stale regardless. */
|
||||
export const STALE_JOB_AGE_MS = 2 * 60 * 60 * 1000;
|
||||
|
||||
/**
|
||||
* How often a running job refreshes its checkpoint between item
|
||||
* completions. Kept well inside the checkpoint window so one missed beat
|
||||
@@ -43,11 +37,14 @@ export const JOB_HEARTBEAT_INTERVAL_MS = 2 * 60 * 1000;
|
||||
export interface InterruptedJobCutoffs {
|
||||
/** Checkpoints (or starts, for jobs without one) before this are too old. */
|
||||
checkpointCutoff: Date;
|
||||
/** Starts before this mark the job stale no matter what. */
|
||||
staleCutoff: Date;
|
||||
}
|
||||
|
||||
export function computeInterruptedJobCutoffs(now: Date = new Date()): InterruptedJobCutoffs {
|
||||
return {
|
||||
checkpointCutoff: new Date(now.getTime() - INTERRUPTED_CHECKPOINT_AGE_MS),
|
||||
staleCutoff: new Date(now.getTime() - STALE_JOB_AGE_MS),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -64,9 +61,10 @@ export interface JobLivenessFields {
|
||||
export function isJobInterrupted(job: JobLivenessFields, now: Date = new Date()): boolean {
|
||||
if (!job.inProgress) return false;
|
||||
|
||||
const { checkpointCutoff } = computeInterruptedJobCutoffs(now);
|
||||
const { checkpointCutoff, staleCutoff } = computeInterruptedJobCutoffs(now);
|
||||
|
||||
if (job.startedAt && job.startedAt < staleCutoff) return true;
|
||||
|
||||
// A checkpoint is the job's own proof of life; its age decides on its own.
|
||||
if (job.lastCheckpoint) return job.lastCheckpoint < checkpointCutoff;
|
||||
|
||||
// No checkpoint yet: give the job the same window it would get after a
|
||||
@@ -84,7 +82,7 @@ export function buildInterruptedJobsCondition(
|
||||
table: typeof mirrorJobs,
|
||||
now: Date = new Date(),
|
||||
): SQL {
|
||||
const { checkpointCutoff } = computeInterruptedJobCutoffs(now);
|
||||
const { checkpointCutoff, staleCutoff } = computeInterruptedJobCutoffs(now);
|
||||
|
||||
return and(
|
||||
eq(table.inProgress, true),
|
||||
@@ -97,6 +95,8 @@ export function buildInterruptedJobsCondition(
|
||||
),
|
||||
// Checkpoint present but too old.
|
||||
lt(table.lastCheckpoint, checkpointCutoff),
|
||||
// Running far too long, whatever the checkpoint says.
|
||||
lt(table.startedAt, staleCutoff),
|
||||
),
|
||||
)!;
|
||||
}
|
||||
@@ -1,101 +0,0 @@
|
||||
import { describe, test, expect } from 'bun:test';
|
||||
import {
|
||||
repositoryIdentityKey,
|
||||
repositoryIdentityKeys,
|
||||
selectNewRepositoriesByIdentity,
|
||||
dedupeRepositoriesByIdentity,
|
||||
} from './repo-utils';
|
||||
|
||||
describe('repositoryIdentityKey', () => {
|
||||
test('is the host plus the full name, independent of the source row', () => {
|
||||
const a = repositoryIdentityKey({ sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools' });
|
||||
const b = repositoryIdentityKey({ sourceProvider: 'github', sourceUrl: 'https://github.com/', normalizedFullName: 'ACME/Tools' });
|
||||
expect(a).toBe(b);
|
||||
});
|
||||
|
||||
test('treats missing provider and url as github.com, like legacy rows', () => {
|
||||
expect(repositoryIdentityKey({ normalizedFullName: 'acme/tools' })).toBe(
|
||||
repositoryIdentityKey({ sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools' })
|
||||
);
|
||||
});
|
||||
|
||||
test('keeps the same name on two hosts apart', () => {
|
||||
const ghe = repositoryIdentityKey({ sourceProvider: 'github', sourceUrl: 'https://ghe.example.com', normalizedFullName: 'acme/tools' });
|
||||
const com = repositoryIdentityKey({ sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools' });
|
||||
expect(ghe).not.toBe(com);
|
||||
});
|
||||
});
|
||||
|
||||
describe('selectNewRepositoriesByIdentity', () => {
|
||||
test('skips a repository already tracked on the same host under another source', () => {
|
||||
const existing = repositoryIdentityKeys([
|
||||
{ sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools' },
|
||||
]);
|
||||
const { fresh, alreadyTracked } = selectNewRepositoriesByIdentity(
|
||||
[
|
||||
{ sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools', sourceId: 'public-only' },
|
||||
{ sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/new', sourceId: 'public-only' },
|
||||
],
|
||||
existing
|
||||
);
|
||||
expect(fresh.map((r) => r.normalizedFullName)).toEqual(['acme/new']);
|
||||
expect(alreadyTracked.map((r) => r.normalizedFullName)).toEqual(['acme/tools']);
|
||||
});
|
||||
|
||||
test('extends the existing set so a later source does not re-add the same repository', () => {
|
||||
const existing = new Set<string>();
|
||||
const first = selectNewRepositoriesByIdentity(
|
||||
[{ sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools' }],
|
||||
existing
|
||||
);
|
||||
const second = selectNewRepositoriesByIdentity(
|
||||
[{ sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools' }],
|
||||
existing
|
||||
);
|
||||
expect(first.fresh).toHaveLength(1);
|
||||
expect(second.fresh).toHaveLength(0);
|
||||
expect(second.alreadyTracked).toHaveLength(1);
|
||||
});
|
||||
|
||||
test('still allows the same name from a different host', () => {
|
||||
const existing = repositoryIdentityKeys([
|
||||
{ sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools' },
|
||||
]);
|
||||
const { fresh } = selectNewRepositoriesByIdentity(
|
||||
[{ sourceProvider: 'gitlab', sourceUrl: 'https://gitlab.com', normalizedFullName: 'acme/tools' }],
|
||||
existing
|
||||
);
|
||||
expect(fresh).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('dedupeRepositoriesByIdentity', () => {
|
||||
test('keeps one row per upstream repository in a batch', () => {
|
||||
const { kept, dropped } = dedupeRepositoriesByIdentity([
|
||||
{ id: 'a', sourceId: 's1', sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools' },
|
||||
{ id: 'b', sourceId: 's2', sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools' },
|
||||
{ id: 'c', sourceId: 's1', sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/other' },
|
||||
]);
|
||||
expect(kept.map((r) => r.id)).toEqual(['a', 'c']);
|
||||
expect(dropped.map((r) => r.id)).toEqual(['b']);
|
||||
});
|
||||
|
||||
test('prefers a row linked to a source over a legacy row without one', () => {
|
||||
const { kept, dropped } = dedupeRepositoriesByIdentity([
|
||||
{ id: 'legacy', sourceId: null, normalizedFullName: 'acme/tools' },
|
||||
{ id: 'linked', sourceId: 's1', sourceProvider: 'github', sourceUrl: 'https://github.com', normalizedFullName: 'acme/tools' },
|
||||
]);
|
||||
expect(kept.map((r) => r.id)).toEqual(['linked']);
|
||||
expect(dropped.map((r) => r.id)).toEqual(['legacy']);
|
||||
});
|
||||
|
||||
test('leaves a batch without duplicates alone', () => {
|
||||
const rows = [
|
||||
{ id: 'a', sourceId: 's1', normalizedFullName: 'acme/a' },
|
||||
{ id: 'b', sourceId: 's1', normalizedFullName: 'acme/b' },
|
||||
];
|
||||
const { kept, dropped } = dedupeRepositoriesByIdentity(rows);
|
||||
expect(kept).toEqual(rows);
|
||||
expect(dropped).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -93,82 +93,3 @@ export function repositoryDestinationColumns(
|
||||
): Pick<RepoInsert, 'destinationProvider' | 'destinationUrl'> {
|
||||
return { destinationProvider: destination.provider, destinationUrl: destination.url };
|
||||
}
|
||||
|
||||
/**
|
||||
* Upstream identity of a repository row: which host it lives on and its full
|
||||
* name there. The repositories table is unique on (userId, sourceId,
|
||||
* normalizedFullName) so that the same owner/name can be tracked on two
|
||||
* hosts, but two source rows can point at the same host (a personal token
|
||||
* source and a public-only source, both github.com), and a legacy row with a
|
||||
* NULL sourceId is distinct from a re-imported one. Discovery used the unique
|
||||
* index alone to decide what was new, so one upstream repository could end up
|
||||
* as two rows that both mirror to the same destination repository and get
|
||||
* processed together in one batch (one of the overlaps behind #417).
|
||||
*/
|
||||
export interface RepositoryIdentityFields {
|
||||
sourceProvider?: string | null;
|
||||
sourceUrl?: string | null;
|
||||
normalizedFullName: string;
|
||||
}
|
||||
|
||||
export function repositoryIdentityKey(row: RepositoryIdentityFields): string {
|
||||
const provider = (row.sourceProvider || 'github').trim().toLowerCase();
|
||||
const url = (row.sourceUrl || 'https://github.com').trim().toLowerCase().replace(/\/+$/, '');
|
||||
return `${provider}|${url}|${row.normalizedFullName.toLowerCase()}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Split discovered rows into the ones not yet tracked for this user on the
|
||||
* same host under any source, and the ones that are. Also drops repeats
|
||||
* within `candidates` themselves. `existing` is extended with every fresh
|
||||
* row, so a caller looping over several sources can pass the same set.
|
||||
*/
|
||||
export function selectNewRepositoriesByIdentity<T extends RepositoryIdentityFields>(
|
||||
candidates: T[],
|
||||
existing: Set<string>
|
||||
): { fresh: T[]; alreadyTracked: T[] } {
|
||||
const fresh: T[] = [];
|
||||
const alreadyTracked: T[] = [];
|
||||
for (const candidate of candidates) {
|
||||
const key = repositoryIdentityKey(candidate);
|
||||
if (existing.has(key)) {
|
||||
alreadyTracked.push(candidate);
|
||||
} else {
|
||||
existing.add(key);
|
||||
fresh.push(candidate);
|
||||
}
|
||||
}
|
||||
return { fresh, alreadyTracked };
|
||||
}
|
||||
|
||||
export function repositoryIdentityKeys(rows: Iterable<RepositoryIdentityFields>): Set<string> {
|
||||
const keys = new Set<string>();
|
||||
for (const row of rows) keys.add(repositoryIdentityKey(row));
|
||||
return keys;
|
||||
}
|
||||
|
||||
/**
|
||||
* Keep one row per upstream identity in a processing batch, so two rows that
|
||||
* resolve to the same destination repository are never mirrored or synced in
|
||||
* the same pass. A row linked to a source wins over a legacy row without one;
|
||||
* otherwise the first row in the batch is kept.
|
||||
*/
|
||||
export function dedupeRepositoriesByIdentity<
|
||||
T extends RepositoryIdentityFields & { sourceId?: string | null },
|
||||
>(rows: T[]): { kept: T[]; dropped: T[] } {
|
||||
const byKey = new Map<string, T>();
|
||||
const dropped: T[] = [];
|
||||
for (const row of rows) {
|
||||
const key = repositoryIdentityKey(row);
|
||||
const current = byKey.get(key);
|
||||
if (!current) {
|
||||
byKey.set(key, row);
|
||||
} else if (!current.sourceId && row.sourceId) {
|
||||
dropped.push(current);
|
||||
byKey.set(key, row);
|
||||
} else {
|
||||
dropped.push(row);
|
||||
}
|
||||
}
|
||||
return { kept: Array.from(byKey.values()), dropped };
|
||||
}
|
||||
@@ -465,34 +465,6 @@ describe.skipIf(!isChild)("Scheduler public-only sources (WP4)", () => {
|
||||
.filter((row) => row.normalizedFullName?.startsWith("pinned-org/"))
|
||||
.map((row) => row.name);
|
||||
expect(orgRepoNamesAfterSecondRun).toEqual(["existing", "newly-published"]);
|
||||
// Nothing new on the second run, so discovery does not even attempt an
|
||||
// insert: already-tracked repositories are filtered out by identity
|
||||
// before the write, not dropped by the unique index afterwards.
|
||||
expect(repoInsertAttempts).toBe(attemptsAfterFirstRun);
|
||||
});
|
||||
|
||||
test("re-discovery does not add a second row for a repository the user already tracks under another source on the same host", async () => {
|
||||
// Imported earlier through a personal token source on github.com, so it
|
||||
// has a different sourceId than the pinned public-only source.
|
||||
repoRows.push({
|
||||
userId: "user-1",
|
||||
sourceId: "source-personal",
|
||||
name: "shared",
|
||||
fullName: "pinned-org/shared",
|
||||
normalizedFullName: "pinned-org/shared",
|
||||
organization: "pinned-org",
|
||||
sourceProvider: "github",
|
||||
sourceUrl: "https://github.com",
|
||||
status: "mirrored",
|
||||
});
|
||||
orgRepoListings.set("pinned-org", [makeGitRepo("shared"), makeGitRepo("fresh")]);
|
||||
|
||||
await schedulerLoop();
|
||||
|
||||
const sharedRows = repoRows.filter((row) => row.normalizedFullName === "pinned-org/shared");
|
||||
expect(sharedRows).toHaveLength(1);
|
||||
expect(sharedRows[0].sourceId).toBe("source-personal");
|
||||
expect(repoRows.some((row) => row.normalizedFullName === "pinned-org/fresh")).toBe(true);
|
||||
expect(consoleLines.some((line) => line.includes("already tracked") && line.includes("pinned-org"))).toBe(true);
|
||||
expect(repoInsertAttempts).toBeGreaterThan(attemptsAfterFirstRun);
|
||||
});
|
||||
});
|
||||
@@ -13,14 +13,7 @@ import { formatDuration } from '@/lib/utils/duration-parser';
|
||||
import type { Repository } from '@/lib/db/schema';
|
||||
import type { Octokit } from '@octokit/rest';
|
||||
import { repoStatusEnum, repositoryVisibilityEnum } from '@/types/Repository';
|
||||
import {
|
||||
mergeGitReposPreferStarred,
|
||||
normalizeGitRepoToInsert,
|
||||
calcBatchSizeForInsert,
|
||||
repositoryIdentityKeys,
|
||||
selectNewRepositoriesByIdentity,
|
||||
dedupeRepositoriesByIdentity,
|
||||
} from '@/lib/repo-utils';
|
||||
import { mergeGitReposPreferStarred, normalizeGitRepoToInsert, calcBatchSizeForInsert } from '@/lib/repo-utils';
|
||||
import { isMirrorableGitHubRepo } from '@/lib/repo-eligibility';
|
||||
import { loadOrganizationForkPolicies, orgForkSkipDecision, resolveOrganizationSkipForks } from '@/lib/utils/mirror-overrides';
|
||||
import { createMirrorJob } from '@/lib/helpers';
|
||||
@@ -109,19 +102,14 @@ async function importRepositoriesFromSources(
|
||||
const orgForkOverrides = await loadOrganizationForkPolicies({ userId });
|
||||
|
||||
const existingRepos = await db
|
||||
.select({
|
||||
normalizedFullName: repositories.normalizedFullName,
|
||||
sourceProvider: repositories.sourceProvider,
|
||||
sourceUrl: repositories.sourceUrl,
|
||||
})
|
||||
.select({ normalizedFullName: repositories.normalizedFullName, sourceId: repositories.sourceId })
|
||||
.from(repositories)
|
||||
.where(eq(repositories.userId, userId));
|
||||
// A repository is identified by its host and full name, not by the source
|
||||
// row that found it. Two sources can point at the same host (a personal
|
||||
// token and a public-only source, both github.com), and a repository the
|
||||
// user already tracks must not be inserted a second time under the other
|
||||
// source: both rows would mirror to the same destination repository.
|
||||
const existingRepoKeys = repositoryIdentityKeys(existingRepos);
|
||||
// Repositories are unique per source: the same full name under two sources
|
||||
// counts as two repositories.
|
||||
const existingRepoKeys = new Set(
|
||||
existingRepos.map(r => `${r.sourceId ?? ''}|${r.normalizedFullName}`)
|
||||
);
|
||||
|
||||
for (const source of sources) {
|
||||
try {
|
||||
@@ -144,21 +132,17 @@ async function importRepositoriesFromSources(
|
||||
const allGithubRepos = mergeGitReposPreferStarred(basicAndForkedRepos, starredRepos);
|
||||
const mirrorableGithubRepos = allGithubRepos.filter(isMirrorableGitHubRepo);
|
||||
|
||||
const candidates = mirrorableGithubRepos.map(repo =>
|
||||
normalizeGitRepoToInsert(repo, { userId, configId: config.id, sourceId: source.id })
|
||||
const newRepos = mirrorableGithubRepos.filter(
|
||||
r => !existingRepoKeys.has(`${source.id}|${r.fullName.toLowerCase()}`)
|
||||
);
|
||||
const { fresh: reposToInsert, alreadyTracked } = selectNewRepositoriesByIdentity(
|
||||
candidates,
|
||||
existingRepoKeys
|
||||
);
|
||||
const newRepos = reposToInsert;
|
||||
if (alreadyTracked.length > 0) {
|
||||
console.log(`[Scheduler] ${alreadyTracked.length} repositories listed by source ${source.name} are already tracked for user ${userId} on the same host; not adding them again`);
|
||||
}
|
||||
|
||||
if (newRepos.length > 0) {
|
||||
console.log(`[Scheduler] Found ${newRepos.length} new repositories for user ${userId} on source ${source.name}`);
|
||||
|
||||
const reposToInsert = newRepos.map(repo =>
|
||||
normalizeGitRepoToInsert(repo, { userId, configId: config.id, sourceId: source.id })
|
||||
);
|
||||
|
||||
// Batch insert to avoid SQLite parameter limit
|
||||
const sample = reposToInsert[0];
|
||||
const columnCount = Object.keys(sample ?? {}).length || 1;
|
||||
@@ -229,19 +213,6 @@ async function rediscoverOrganizationRepositories(
|
||||
)
|
||||
);
|
||||
|
||||
// Same identity rule as personal discovery: an organization pinned to a
|
||||
// public-only source must not re-insert repositories the personal source
|
||||
// already tracks on the same host (or the other way round).
|
||||
const existingRepos = await db
|
||||
.select({
|
||||
normalizedFullName: repositories.normalizedFullName,
|
||||
sourceProvider: repositories.sourceProvider,
|
||||
sourceUrl: repositories.sourceUrl,
|
||||
})
|
||||
.from(repositories)
|
||||
.where(eq(repositories.userId, userId));
|
||||
const existingRepoKeys = repositoryIdentityKeys(existingRepos);
|
||||
|
||||
for (const org of includedOrgs) {
|
||||
try {
|
||||
const source = findSourceForOrganization(org, sources);
|
||||
@@ -257,17 +228,14 @@ async function rediscoverOrganizationRepositories(
|
||||
repo => repo.isDisabled !== true && !(skipOrgForks && repo.isForked)
|
||||
);
|
||||
|
||||
const { fresh: repoRecords, alreadyTracked } = selectNewRepositoriesByIdentity(
|
||||
mirrorableRepos.map(repo =>
|
||||
if (mirrorableRepos.length > 0) {
|
||||
const repoRecords = mirrorableRepos.map(repo =>
|
||||
normalizeGitRepoToInsert(
|
||||
{ ...repo, organization: repo.organization ?? org.name },
|
||||
{ userId, configId: config.id, sourceId: source.id }
|
||||
)
|
||||
),
|
||||
existingRepoKeys
|
||||
);
|
||||
);
|
||||
|
||||
if (repoRecords.length > 0) {
|
||||
// Batch insert to avoid SQLite parameter limit
|
||||
const sample = repoRecords[0];
|
||||
const columnCount = Object.keys(sample ?? {}).length || 1;
|
||||
@@ -279,10 +247,7 @@ async function rediscoverOrganizationRepositories(
|
||||
.values(batch)
|
||||
.onConflictDoNothing({ target: [repositories.userId, repositories.sourceId, repositories.normalizedFullName] });
|
||||
}
|
||||
console.log(`[Scheduler] Re-discovered ${repoRecords.length} new repositories for organization ${org.name} on source ${source.name} for user ${userId} during ${phase}`);
|
||||
}
|
||||
if (alreadyTracked.length > 0) {
|
||||
console.log(`[Scheduler] Organization ${org.name}: ${alreadyTracked.length} repositories are already tracked for user ${userId} on the same host, not adding them again under source ${source.name}`);
|
||||
console.log(`[Scheduler] Re-discovered ${mirrorableRepos.length} repositories for organization ${org.name} on source ${source.name} for user ${userId} during ${phase}`);
|
||||
}
|
||||
|
||||
await db
|
||||
@@ -366,24 +331,6 @@ function sourceUsernamesBySourceId(sources: SourceRecord[]): Map<string, string>
|
||||
/**
|
||||
* Run scheduled mirror sync for a single user configuration
|
||||
*/
|
||||
/**
|
||||
* Keep one row per upstream repository in a scheduler batch. Duplicate rows
|
||||
* (the same repository imported under two sources on the same host) resolve
|
||||
* to one destination repository, and processing both in one Promise.all ran
|
||||
* every step twice against it. The dropped rows are logged so the duplicate
|
||||
* can be cleaned up; discovery no longer creates new ones.
|
||||
*/
|
||||
function dropDuplicateRowsForPass<T extends { fullName: string; sourceId?: string | null; sourceProvider?: string | null; sourceUrl?: string | null; normalizedFullName: string }>(
|
||||
rows: T[],
|
||||
pass: string
|
||||
): T[] {
|
||||
const { kept, dropped } = dedupeRepositoriesByIdentity(rows);
|
||||
for (const row of dropped) {
|
||||
console.warn(`[Scheduler] ${row.fullName} is tracked more than once (another row on the same host covers it); skipping the duplicate row in this ${pass} pass`);
|
||||
}
|
||||
return kept;
|
||||
}
|
||||
|
||||
async function runScheduledSync(config: any): Promise<void> {
|
||||
const userId = config.userId;
|
||||
console.log(`[Scheduler] Running scheduled sync for user ${userId}`);
|
||||
@@ -510,11 +457,6 @@ async function runScheduledSync(config: any): Promise<void> {
|
||||
console.log(`[Scheduler] Skipped ${forkSkippedCount - reposNeedingMirror.length} forked repositories from auto-mirror (organization skip forks)`);
|
||||
}
|
||||
|
||||
// Two rows for one upstream repository (imported under two sources
|
||||
// that point at the same host) resolve to one destination
|
||||
// repository; mirror it once per pass, not twice at the same time.
|
||||
reposNeedingMirror = dropDuplicateRowsForPass(reposNeedingMirror, 'auto-mirror');
|
||||
|
||||
if (reposNeedingMirror.length > 0) {
|
||||
console.log(`[Scheduler] Found ${reposNeedingMirror.length} repositories that need initial mirroring`);
|
||||
|
||||
@@ -642,8 +584,6 @@ async function runScheduledSync(config: any): Promise<void> {
|
||||
// Non-critical logging, ignore errors
|
||||
}
|
||||
|
||||
reposToSync = dropDuplicateRowsForPass(reposToSync, 'sync');
|
||||
|
||||
if (reposToSync.length === 0) {
|
||||
console.log(`[Scheduler] No repositories to sync for user ${userId}`);
|
||||
return;
|
||||
@@ -868,7 +808,6 @@ async function performInitialAutoStart(): Promise<void> {
|
||||
if (skippedCount > 0) {
|
||||
console.log(`[Scheduler] Skipped ${skippedCount} repositories from initial auto-mirror (autoMirror=${autoMirrorOwned}, autoMirrorStarred=${autoMirrorStarred})`);
|
||||
}
|
||||
reposNeedingMirror = dropDuplicateRowsForPass(reposNeedingMirror, 'initial auto-mirror');
|
||||
|
||||
if (reposNeedingMirror.length > 0) {
|
||||
console.log(`[Scheduler] Found ${reposNeedingMirror.length} repositories that need mirroring`);
|
||||
|
||||
@@ -165,46 +165,3 @@ describe("processWithRetry", () => {
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe("processWithRetry checkpointing", () => {
|
||||
test("records every completed item and publishes on the interval", async () => {
|
||||
const items = ["a", "b", "c", "d", "e"];
|
||||
const calls: Array<[string, string, boolean]> = [];
|
||||
|
||||
await processWithRetry(items, async (item) => item.toUpperCase(), {
|
||||
concurrencyLimit: 1,
|
||||
jobId: "job-1",
|
||||
getItemId: (item) => item,
|
||||
checkpointInterval: 2,
|
||||
onCheckpoint: async (jobId, itemId, publish) => {
|
||||
calls.push([jobId, itemId, publish]);
|
||||
},
|
||||
});
|
||||
|
||||
// One checkpoint per item, in completion order, and no "final" placeholder.
|
||||
expect(calls.map(([, itemId]) => itemId)).toEqual(items);
|
||||
expect(calls.every(([jobId]) => jobId === "job-1")).toBe(true);
|
||||
// Progress events only every second item.
|
||||
expect(calls.map(([, , publish]) => publish)).toEqual([false, true, false, true, false]);
|
||||
});
|
||||
|
||||
test("does not checkpoint an item that failed for good", async () => {
|
||||
const calls: string[] = [];
|
||||
|
||||
// processInParallel logs a failed item and carries on, so the call resolves.
|
||||
await processWithRetry(["ok", "bad"], async (item) => {
|
||||
if (item === "bad") throw new Error("nope");
|
||||
return item;
|
||||
}, {
|
||||
concurrencyLimit: 1,
|
||||
maxRetries: 0,
|
||||
jobId: "job-2",
|
||||
getItemId: (item) => item,
|
||||
onCheckpoint: async (_jobId, itemId) => {
|
||||
calls.push(itemId);
|
||||
},
|
||||
});
|
||||
|
||||
expect(calls).toEqual(["ok"]);
|
||||
});
|
||||
});
|
||||
@@ -93,15 +93,8 @@ export async function processWithRetry<T, R>(
|
||||
onRetry?: (item: T, error: Error, attempt: number) => void;
|
||||
jobId?: string; // Optional job ID for checkpointing
|
||||
getItemId?: (item: T) => string; // Function to get a unique ID for each item
|
||||
/**
|
||||
* Called once for EVERY completed item, so a resume knows exactly what is
|
||||
* done. `publish` is true every `checkpointInterval` items: the callback
|
||||
* uses it to throttle progress events, not the checkpoint itself. It used
|
||||
* to be called only for the item that tripped the interval, and a resume
|
||||
* re-processed everything finished since the previous checkpoint.
|
||||
*/
|
||||
onCheckpoint?: (jobId: string, completedItemId: string, publish: boolean) => Promise<void>;
|
||||
checkpointInterval?: number; // How many completed items between published progress events
|
||||
onCheckpoint?: (jobId: string, completedItemId: string) => Promise<void>; // Callback for checkpointing
|
||||
checkpointInterval?: number; // How many items to process before checkpointing
|
||||
} = {}
|
||||
): Promise<R[]> {
|
||||
const {
|
||||
@@ -137,16 +130,16 @@ export async function processWithRetry<T, R>(
|
||||
|
||||
const result = await processItem(item);
|
||||
|
||||
// Record every completed item; publish progress on the interval.
|
||||
// Handle checkpointing if enabled
|
||||
if (jobId && getItemId && onCheckpoint) {
|
||||
const itemId = getItemId(item);
|
||||
itemsProcessedSinceLastCheckpoint++;
|
||||
|
||||
const publish = itemsProcessedSinceLastCheckpoint >= checkpointInterval;
|
||||
if (publish) {
|
||||
// Checkpoint based on the interval
|
||||
if (itemsProcessedSinceLastCheckpoint >= checkpointInterval) {
|
||||
await onCheckpoint(jobId, itemId);
|
||||
itemsProcessedSinceLastCheckpoint = 0;
|
||||
}
|
||||
await onCheckpoint(jobId, itemId, publish);
|
||||
}
|
||||
|
||||
return result;
|
||||
@@ -193,6 +186,12 @@ export async function processWithRetry<T, R>(
|
||||
onProgress
|
||||
);
|
||||
|
||||
// Final checkpoint if there are remaining items since the last checkpoint
|
||||
if (jobId && getItemId && onCheckpoint && itemsProcessedSinceLastCheckpoint > 0) {
|
||||
// We don't have a specific item ID for the final checkpoint, so we'll use a placeholder
|
||||
await onCheckpoint(jobId, 'final');
|
||||
}
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
@@ -310,10 +309,8 @@ export async function processWithResilience<T, R>(
|
||||
});
|
||||
}, JOB_HEARTBEAT_INTERVAL_MS);
|
||||
|
||||
// Define the checkpoint function. Every completed item is written to the
|
||||
// job row so a resume skips exactly what is done; only every
|
||||
// checkpointInterval-th write also publishes a progress event.
|
||||
const onCheckpoint = async (jobId: string, completedItemId: string, publish: boolean = true) => {
|
||||
// Define the checkpoint function
|
||||
const onCheckpoint = async (jobId: string, completedItemId: string) => {
|
||||
const itemName = items.find(item => getItemId(item) === completedItemId)
|
||||
? getItemName(items.find(item => getItemId(item) === completedItemId)!)
|
||||
: 'unknown';
|
||||
@@ -322,7 +319,6 @@ export async function processWithResilience<T, R>(
|
||||
jobId,
|
||||
completedItemId,
|
||||
message: `Processed item: ${itemName}`,
|
||||
silent: !publish,
|
||||
});
|
||||
};
|
||||
|
||||
|
||||
+10
-16
@@ -9,8 +9,6 @@ import {
|
||||
mergeGitReposPreferStarred,
|
||||
normalizeGitRepoToInsert,
|
||||
calcBatchSizeForInsert,
|
||||
repositoryIdentityKeys,
|
||||
selectNewRepositoriesByIdentity,
|
||||
} from "@/lib/repo-utils";
|
||||
import { loadOrganizationForkPolicies } from "@/lib/utils/mirror-overrides";
|
||||
import { requireAuthenticatedUserId } from "@/lib/auth-guards";
|
||||
@@ -167,8 +165,7 @@ export const POST: APIRoute = async ({ request, locals }) => {
|
||||
tx
|
||||
.select({
|
||||
normalizedFullName: repositories.normalizedFullName,
|
||||
sourceProvider: repositories.sourceProvider,
|
||||
sourceUrl: repositories.sourceUrl,
|
||||
sourceId: repositories.sourceId,
|
||||
})
|
||||
.from(repositories)
|
||||
.where(eq(repositories.userId, userId)),
|
||||
@@ -178,20 +175,17 @@ export const POST: APIRoute = async ({ request, locals }) => {
|
||||
.where(eq(organizations.userId, userId)),
|
||||
]);
|
||||
|
||||
// A repository is identified by its host and full name, whichever
|
||||
// source row lists it. The same name on two hosts is two
|
||||
// repositories; the same repository seen by two sources on one host
|
||||
// (a personal token and a public-only source) is one, and must not
|
||||
// become two rows that mirror to the same destination.
|
||||
const existingRepoKeys = repositoryIdentityKeys(existingRepos);
|
||||
// The same full name under two sources is two different repositories.
|
||||
const existingRepoKeys = new Set(
|
||||
existingRepos.map((r) => `${r.sourceId ?? ""}|${r.normalizedFullName}`)
|
||||
);
|
||||
const existingOrgMap = new Map(existingOrgs.map((o) => [o.normalizedName, o.status]));
|
||||
|
||||
insertedRepos = selectNewRepositoriesByIdentity(
|
||||
newRepos.filter(
|
||||
(r) => !r.organization || !ignoredOrgNames.has(r.organization.toLowerCase())
|
||||
),
|
||||
existingRepoKeys
|
||||
).fresh;
|
||||
insertedRepos = newRepos.filter(
|
||||
(r) =>
|
||||
!existingRepoKeys.has(`${source.id}|${r.normalizedFullName}`) &&
|
||||
(!r.organization || !ignoredOrgNames.has(r.organization.toLowerCase()))
|
||||
);
|
||||
insertedOrgs = newOrgs.filter((o) => !existingOrgMap.has(o.normalizedName));
|
||||
|
||||
// Update previously failed orgs that now succeeded. The pin must
|
||||
|
||||
@@ -148,17 +148,6 @@ if (isChild) {
|
||||
select: () => ({
|
||||
from: (table: unknown) => ({
|
||||
where: (cond: unknown) => ({
|
||||
// Awaited directly by the existing-repository lookup that keeps
|
||||
// one row per upstream repository; the inserted rows stand in
|
||||
// for the table.
|
||||
then: (
|
||||
onfulfilled: (rows: OrgRow[]) => unknown,
|
||||
onrejected?: (reason: unknown) => unknown
|
||||
) =>
|
||||
Promise.resolve(table === repositoriesTable ? [...insertedRepoRows] : []).then(
|
||||
onfulfilled,
|
||||
onrejected
|
||||
),
|
||||
limit: async (): Promise<OrgRow[]> => {
|
||||
const leaves = conditionLeafValues(cond);
|
||||
if (table === organizationsTable) {
|
||||
|
||||
@@ -15,12 +15,7 @@ import {
|
||||
normalizeSourceUrl,
|
||||
} from "@/lib/source-providers/kinds";
|
||||
import type { SourceRecord } from "@/lib/sources";
|
||||
import {
|
||||
normalizeGitRepoToInsert,
|
||||
calcBatchSizeForInsert,
|
||||
repositoryIdentityKeys,
|
||||
selectNewRepositoriesByIdentity,
|
||||
} from "@/lib/repo-utils";
|
||||
import { normalizeGitRepoToInsert, calcBatchSizeForInsert } from "@/lib/repo-utils";
|
||||
import { resolveOrganizationSkipForks } from "@/lib/utils/mirror-overrides";
|
||||
import { requireAuthenticatedUserId } from "@/lib/auth-guards";
|
||||
|
||||
@@ -264,31 +259,12 @@ export const POST: APIRoute = async ({ request, locals }) => {
|
||||
);
|
||||
|
||||
// Insert repositories. The normalizer stamps the source provider and URL.
|
||||
// Repositories the user already tracks on this host under another source
|
||||
// (the personal source, say) are not inserted a second time: two rows for
|
||||
// one upstream repository would both mirror to the same destination.
|
||||
const existingRepos = await db
|
||||
.select({
|
||||
normalizedFullName: repositories.normalizedFullName,
|
||||
sourceProvider: repositories.sourceProvider,
|
||||
sourceUrl: repositories.sourceUrl,
|
||||
})
|
||||
.from(repositories)
|
||||
.where(eq(repositories.userId, userId));
|
||||
const { fresh: repoRecords, alreadyTracked } = selectNewRepositoriesByIdentity(
|
||||
mirrorableRepos.map((repo) =>
|
||||
normalizeGitRepoToInsert(
|
||||
{ ...repo, organization: repo.organization ?? orgData.name },
|
||||
{ userId, configId, sourceId: source.id }
|
||||
)
|
||||
),
|
||||
repositoryIdentityKeys(existingRepos)
|
||||
const repoRecords = mirrorableRepos.map((repo) =>
|
||||
normalizeGitRepoToInsert(
|
||||
{ ...repo, organization: repo.organization ?? orgData.name },
|
||||
{ userId, configId, sourceId: source.id }
|
||||
)
|
||||
);
|
||||
if (alreadyTracked.length > 0) {
|
||||
console.log(
|
||||
`[Organization import] ${alreadyTracked.length} repositories of ${orgData.name} are already tracked for user ${userId} on this host; not adding them again under source ${source.name}`
|
||||
);
|
||||
}
|
||||
|
||||
// Batch insert repositories to avoid SQLite parameter limit
|
||||
// Compute batch size based on column count
|
||||
|
||||
+1
-2
@@ -38,8 +38,7 @@
|
||||
"esbuild@>=0.27.3 <0.28.1": ">=0.28.1",
|
||||
"sharp": "^0.35.4",
|
||||
"svgo": "^4.1.0",
|
||||
"js-yaml": "^4.3.2",
|
||||
"smol-toml": "^1.7.1"
|
||||
"js-yaml": "^4.3.2"
|
||||
}
|
||||
}
|
||||
}
|
||||
Generated
+7
-8
@@ -9,7 +9,6 @@ overrides:
|
||||
sharp: ^0.35.4
|
||||
svgo: ^4.1.0
|
||||
js-yaml: ^4.3.2
|
||||
smol-toml: ^1.7.1
|
||||
|
||||
importers:
|
||||
|
||||
@@ -2023,8 +2022,8 @@ packages:
|
||||
engines: {node: '>=20.19.5', npm: '>=10.8.2'}
|
||||
hasBin: true
|
||||
|
||||
smol-toml@1.8.0:
|
||||
resolution: {integrity: sha512-kCZr2V3ch9i00x8zXRhjUNVcjG9ijES5dDudkXvUVCT5QlJNQWElSJdZqyPemffHoLNUYwOcou0Fy+ojN0uHSQ==}
|
||||
smol-toml@1.7.0:
|
||||
resolution: {integrity: sha512-aqVvWoyO21L23mb+drl4RmMXbf6N7FdHjAhTRA9ZBL7apWBgfWC16KjrASI+1p9GAroljyMHj6fK67i0UiTNvQ==}
|
||||
engines: {node: '>= 18'}
|
||||
|
||||
source-map-js@1.2.1:
|
||||
@@ -2365,7 +2364,7 @@ snapshots:
|
||||
picomatch: 4.0.5
|
||||
retext-smartypants: 6.2.0
|
||||
shiki: 4.3.1
|
||||
smol-toml: 1.8.0
|
||||
smol-toml: 1.7.0
|
||||
unified: 11.0.5
|
||||
|
||||
'@astrojs/internal-helpers@0.11.0':
|
||||
@@ -2376,7 +2375,7 @@ snapshots:
|
||||
picomatch: 4.0.5
|
||||
retext-smartypants: 6.2.0
|
||||
shiki: 4.3.1
|
||||
smol-toml: 1.8.0
|
||||
smol-toml: 1.7.0
|
||||
unified: 11.0.5
|
||||
|
||||
'@astrojs/markdown-remark@7.2.1':
|
||||
@@ -2641,7 +2640,7 @@ snapshots:
|
||||
dependencies:
|
||||
'@emnapi/core': 1.11.1
|
||||
'@emnapi/runtime': 1.11.1
|
||||
'@napi-rs/wasm-runtime': 1.2.3(@emnapi/core@1.11.1)(@emnapi/runtime@1.11.1)
|
||||
'@napi-rs/wasm-runtime': 1.1.6(@emnapi/core@1.11.1)(@emnapi/runtime@1.11.1)
|
||||
optional: true
|
||||
|
||||
'@bruits/satteri-win32-arm64-msvc@0.10.5':
|
||||
@@ -3296,7 +3295,7 @@ snapshots:
|
||||
picomatch: 4.0.5
|
||||
semver: 7.8.5
|
||||
shiki: 4.3.1
|
||||
smol-toml: 1.8.0
|
||||
smol-toml: 1.7.0
|
||||
svgo: 4.1.0
|
||||
tinyclip: 0.1.15
|
||||
tinyexec: 1.2.4
|
||||
@@ -4681,7 +4680,7 @@ snapshots:
|
||||
arg: 5.0.2
|
||||
sax: 1.6.0
|
||||
|
||||
smol-toml@1.8.0: {}
|
||||
smol-toml@1.7.0: {}
|
||||
|
||||
source-map-js@1.2.1: {}
|
||||
|
||||
|
||||
Reference in new issue
Block a user