fix: limit concurrent ls to 2 in flight calls
This commit is contained in:
parent
34fc4f0cbb
commit
5520d6ca17
4 changed files with 103 additions and 15 deletions
|
|
@ -5,7 +5,6 @@ import { FileBrowser, type FileBrowserUiProps } from "~/client/components/file-b
|
|||
import { useFileBrowser } from "~/client/hooks/use-file-browser";
|
||||
import { parseError } from "~/client/lib/errors";
|
||||
import { isPathWithin, normalizeAbsolutePath } from "@zerobyte/core/utils";
|
||||
import { logger } from "~/client/lib/logger";
|
||||
|
||||
function createPathPrefixFns(basePath: string) {
|
||||
return {
|
||||
|
|
@ -84,16 +83,6 @@ export const SnapshotTreeBrowser = (props: SnapshotTreeBrowserProps) => {
|
|||
}),
|
||||
);
|
||||
},
|
||||
prefetchFolder: (displayPath) => {
|
||||
void queryClient
|
||||
.prefetchQuery(
|
||||
listSnapshotFilesOptions({
|
||||
path: { shortId: repositoryId, snapshotId },
|
||||
query: { path: displayPath, offset: 0, limit: pageSize },
|
||||
}),
|
||||
)
|
||||
.catch((e) => logger.error(e));
|
||||
},
|
||||
pathTransform: displayPathFns,
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -63,9 +63,7 @@ export function ScheduleDetailsPage(props: Props) {
|
|||
const queryClient = useQueryClient();
|
||||
const navigate = useNavigate();
|
||||
const searchParams = useSearch({ from: "/(dashboard)/backups/$backupId/" });
|
||||
const [selectedSnapshotId, setSelectedSnapshotId] = useState<string | undefined>(
|
||||
initialSnapshotId ?? loaderData.snapshots?.at(-1)?.short_id,
|
||||
);
|
||||
const [selectedSnapshotId, setSelectedSnapshotId] = useState<string | undefined>(initialSnapshotId);
|
||||
const [showDeleteConfirm, setShowDeleteConfirm] = useState(false);
|
||||
const [snapshotToDelete, setSnapshotToDelete] = useState<string | null>(null);
|
||||
|
||||
|
|
|
|||
|
|
@ -208,6 +208,95 @@ describe("repositoriesService repository stats", () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe("repositoriesService.listSnapshotFiles", () => {
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
test("limits concurrent restic ls commands per repository", async () => {
|
||||
const repository = await createTestRepository(session.organizationId);
|
||||
let active = 0;
|
||||
let maxActive = 0;
|
||||
let releaseAll = false;
|
||||
const releaseWaiters: Array<() => void> = [];
|
||||
|
||||
const releaseWaitingCommands = () => {
|
||||
const waiters = releaseWaiters.splice(0);
|
||||
for (const release of waiters) {
|
||||
release();
|
||||
}
|
||||
};
|
||||
|
||||
const lsSpy = vi.spyOn(restic, "ls").mockImplementation((_config, snapshotId, _path, options) =>
|
||||
Effect.promise(async () => {
|
||||
active++;
|
||||
maxActive = Math.max(maxActive, active);
|
||||
|
||||
try {
|
||||
if (!releaseAll) {
|
||||
await new Promise<void>((resolve) => releaseWaiters.push(resolve));
|
||||
}
|
||||
|
||||
return {
|
||||
snapshot: {
|
||||
id: snapshotId,
|
||||
short_id: snapshotId,
|
||||
time: new Date().toISOString(),
|
||||
tree: "tree",
|
||||
paths: ["/"],
|
||||
hostname: "host",
|
||||
struct_type: "snapshot" as const,
|
||||
message_type: "snapshot" as const,
|
||||
},
|
||||
nodes: [],
|
||||
pagination: {
|
||||
offset: options.offset ?? 0,
|
||||
limit: options.limit ?? 500,
|
||||
total: 0,
|
||||
hasMore: false,
|
||||
},
|
||||
};
|
||||
} finally {
|
||||
active--;
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
const calls = Array.from({ length: 4 }, (_, index) =>
|
||||
withContext({ organizationId: session.organizationId, userId: session.user.id }, () =>
|
||||
repositoriesService.listSnapshotFiles(repository.shortId, `snapshot-${index}`, "/", {
|
||||
offset: 0,
|
||||
limit: 100,
|
||||
}),
|
||||
),
|
||||
);
|
||||
|
||||
try {
|
||||
await waitForExpect(() => {
|
||||
expect(releaseWaiters).toHaveLength(2);
|
||||
});
|
||||
expect(maxActive).toBe(2);
|
||||
|
||||
releaseWaitingCommands();
|
||||
|
||||
await waitForExpect(() => {
|
||||
expect(releaseWaiters).toHaveLength(2);
|
||||
});
|
||||
expect(maxActive).toBe(2);
|
||||
|
||||
releaseWaitingCommands();
|
||||
await Promise.all(calls);
|
||||
} finally {
|
||||
releaseAll = true;
|
||||
releaseWaitingCommands();
|
||||
await Promise.allSettled(calls);
|
||||
}
|
||||
|
||||
expect(lsSpy).toHaveBeenCalledTimes(4);
|
||||
expect(maxActive).toBeLessThanOrEqual(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe("repositoriesService.dumpSnapshot", () => {
|
||||
afterEach(() => {
|
||||
vi.restoreAllMocks();
|
||||
|
|
|
|||
|
|
@ -42,6 +42,7 @@ import type { ParsedTask, TaskInput } from "../tasks/tasks.schemas";
|
|||
import { Effect } from "effect";
|
||||
|
||||
const runningDoctors = new Map<string, AbortController>();
|
||||
const lsLimiters = new Map<string, Effect.Semaphore>();
|
||||
const RESTORE_TASK_RESOURCE_TYPE = "repository";
|
||||
|
||||
type RestoreTaskInput = Extract<TaskInput, { kind: "restore" }>;
|
||||
|
|
@ -99,6 +100,15 @@ const updateActiveRestoreTask = (restoreId: string, eventName: string, update: (
|
|||
}
|
||||
};
|
||||
|
||||
const getLsLimiter = (repositoryId: string) => {
|
||||
let limiter = lsLimiters.get(repositoryId);
|
||||
if (!limiter) {
|
||||
limiter = Effect.unsafeMakeSemaphore(2);
|
||||
lsLimiters.set(repositoryId, limiter);
|
||||
}
|
||||
return limiter;
|
||||
};
|
||||
|
||||
const findActiveRestoreTask = (
|
||||
organizationId: string,
|
||||
repositoryShortId: string,
|
||||
|
|
@ -472,7 +482,9 @@ const listSnapshotFiles = async (
|
|||
const releaseLock = await repoMutex.acquireShared(repository.id, `ls:${snapshotId}`);
|
||||
try {
|
||||
const result = await runEffectPromise(
|
||||
restic.ls(repository.config, snapshotId, path, { organizationId, offset, limit }),
|
||||
restic
|
||||
.ls(repository.config, snapshotId, path, { organizationId, offset, limit })
|
||||
.pipe(getLsLimiter(repository.id).withPermits(1)),
|
||||
);
|
||||
|
||||
if (!result.snapshot) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue