427 lines
11 KiB
TypeScript
427 lines
11 KiB
TypeScript
import { logger } from "@zerobyte/core/node";
|
|
|
|
export type LockType = "shared" | "exclusive";
|
|
|
|
export interface LockRequest {
|
|
repositoryId: string;
|
|
type: LockType;
|
|
operation: string;
|
|
}
|
|
|
|
interface LockHolder {
|
|
id: string;
|
|
operation: string;
|
|
acquiredAt: number;
|
|
}
|
|
|
|
interface RepositoryLockState {
|
|
sharedHolders: Map<string, LockHolder>;
|
|
exclusiveHolder: LockHolder | null;
|
|
waitQueue: Array<{
|
|
type: LockType;
|
|
operation: string;
|
|
resolve: (lockId: string) => void;
|
|
}>;
|
|
}
|
|
|
|
class RepositoryMutex {
|
|
private locks = new Map<string, RepositoryLockState>();
|
|
private changeListeners = new Map<string, Set<() => void>>();
|
|
private lockIdCounter = 0;
|
|
|
|
private getOrCreateState(repositoryId: string): RepositoryLockState {
|
|
let state = this.locks.get(repositoryId);
|
|
if (!state) {
|
|
state = {
|
|
sharedHolders: new Map(),
|
|
exclusiveHolder: null,
|
|
waitQueue: [],
|
|
};
|
|
this.locks.set(repositoryId, state);
|
|
}
|
|
return state;
|
|
}
|
|
|
|
private generateLockId(): string {
|
|
return `lock_${++this.lockIdCounter}_${Date.now()}`;
|
|
}
|
|
|
|
private cleanupStateIfEmpty(repositoryId: string): void {
|
|
const state = this.locks.get(repositoryId);
|
|
if (state && state.sharedHolders.size === 0 && !state.exclusiveHolder && state.waitQueue.length === 0) {
|
|
this.locks.delete(repositoryId);
|
|
}
|
|
}
|
|
|
|
private notifyChange(repositoryId: string): void {
|
|
const listeners = this.changeListeners.get(repositoryId);
|
|
if (!listeners) {
|
|
return;
|
|
}
|
|
|
|
for (const listener of listeners) {
|
|
listener();
|
|
}
|
|
}
|
|
|
|
private canAcquireImmediately(state: RepositoryLockState | undefined, type: LockType): boolean {
|
|
if (!state) {
|
|
return true;
|
|
}
|
|
|
|
if (type === "shared") {
|
|
const hasExclusiveInQueue = state.waitQueue.some((item) => item.type === "exclusive");
|
|
return !state.exclusiveHolder && !hasExclusiveInQueue;
|
|
}
|
|
|
|
return !state.exclusiveHolder && state.sharedHolders.size === 0 && state.waitQueue.length === 0;
|
|
}
|
|
|
|
private waitForChange(repositoryIds: string[], signal?: AbortSignal) {
|
|
if (signal?.aborted) {
|
|
throw signal.reason || new Error("Operation aborted");
|
|
}
|
|
|
|
return new Promise<void>((resolve, reject) => {
|
|
const uniqueRepositoryIds = [...new Set(repositoryIds)];
|
|
const cleanupCallbacks: Array<() => void> = [];
|
|
let settled = false;
|
|
|
|
const settle = (callback: () => void) => {
|
|
if (settled) {
|
|
return;
|
|
}
|
|
|
|
settled = true;
|
|
for (const cleanup of cleanupCallbacks) {
|
|
cleanup();
|
|
}
|
|
callback();
|
|
};
|
|
|
|
for (const repositoryId of uniqueRepositoryIds) {
|
|
let listeners = this.changeListeners.get(repositoryId);
|
|
if (!listeners) {
|
|
listeners = new Set();
|
|
this.changeListeners.set(repositoryId, listeners);
|
|
}
|
|
|
|
const listener = () => settle(resolve);
|
|
listeners.add(listener);
|
|
|
|
cleanupCallbacks.push(() => {
|
|
const currentListeners = this.changeListeners.get(repositoryId);
|
|
if (!currentListeners) {
|
|
return;
|
|
}
|
|
|
|
currentListeners.delete(listener);
|
|
if (currentListeners.size === 0) {
|
|
this.changeListeners.delete(repositoryId);
|
|
}
|
|
});
|
|
}
|
|
|
|
if (signal) {
|
|
const onAbort = () => {
|
|
settle(() => reject(signal.reason || new Error("Operation aborted")));
|
|
};
|
|
|
|
signal.addEventListener("abort", onAbort);
|
|
cleanupCallbacks.push(() => {
|
|
signal.removeEventListener("abort", onAbort);
|
|
});
|
|
}
|
|
});
|
|
}
|
|
|
|
private tryAcquireMany(requests: LockRequest[]) {
|
|
for (const request of requests) {
|
|
if (!this.canAcquireImmediately(this.locks.get(request.repositoryId), request.type)) {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
const releases = requests.map((request) => {
|
|
const state = this.getOrCreateState(request.repositoryId);
|
|
const lockId = this.generateLockId();
|
|
|
|
if (request.type === "shared") {
|
|
state.sharedHolders.set(lockId, {
|
|
id: lockId,
|
|
operation: request.operation,
|
|
acquiredAt: Date.now(),
|
|
});
|
|
} else {
|
|
state.exclusiveHolder = {
|
|
id: lockId,
|
|
operation: request.operation,
|
|
acquiredAt: Date.now(),
|
|
};
|
|
}
|
|
|
|
return this.createRelease(request.type, request.repositoryId, lockId);
|
|
});
|
|
|
|
let released = false;
|
|
return () => {
|
|
if (released) {
|
|
return;
|
|
}
|
|
|
|
released = true;
|
|
for (const release of releases.toReversed()) {
|
|
release();
|
|
}
|
|
};
|
|
}
|
|
|
|
async acquireShared(repositoryId: string, operation: string, signal?: AbortSignal): Promise<() => void> {
|
|
if (signal?.aborted) {
|
|
throw signal.reason || new Error("Operation aborted");
|
|
}
|
|
|
|
const state = this.getOrCreateState(repositoryId);
|
|
|
|
const hasExclusiveInQueue = state.waitQueue.some((item) => item.type === "exclusive");
|
|
|
|
if (!state.exclusiveHolder && !hasExclusiveInQueue) {
|
|
const lockId = this.generateLockId();
|
|
state.sharedHolders.set(lockId, {
|
|
id: lockId,
|
|
operation,
|
|
acquiredAt: Date.now(),
|
|
});
|
|
return () => this.releaseShared(repositoryId, lockId);
|
|
}
|
|
|
|
logger.debug(
|
|
`[Mutex] Waiting for shared lock on repo ${repositoryId}: ${operation} (exclusive held by: ${state.exclusiveHolder?.operation ?? "none"}, queue: ${state.waitQueue.length})`,
|
|
);
|
|
|
|
let onAbort: () => void = () => {};
|
|
let lockId: string | undefined;
|
|
try {
|
|
lockId = await new Promise<string>((resolve, reject) => {
|
|
const waiter = { type: "shared" as const, operation, resolve };
|
|
state.waitQueue.push(waiter);
|
|
|
|
if (signal) {
|
|
onAbort = () => {
|
|
const index = state.waitQueue.indexOf(waiter);
|
|
if (index !== -1) {
|
|
state.waitQueue.splice(index, 1);
|
|
this.cleanupStateIfEmpty(repositoryId);
|
|
this.notifyChange(repositoryId);
|
|
reject(signal.reason || new Error("Operation aborted"));
|
|
}
|
|
};
|
|
signal.addEventListener("abort", onAbort);
|
|
}
|
|
});
|
|
} finally {
|
|
signal?.removeEventListener("abort", onAbort);
|
|
}
|
|
|
|
if (signal?.aborted) {
|
|
if (lockId) {
|
|
this.releaseShared(repositoryId, lockId);
|
|
}
|
|
throw signal.reason || new Error("Operation aborted");
|
|
}
|
|
|
|
return this.createRelease("shared", repositoryId, lockId!);
|
|
}
|
|
|
|
async acquireExclusive(repositoryId: string, operation: string, signal?: AbortSignal): Promise<() => void> {
|
|
if (signal?.aborted) {
|
|
throw signal.reason || new Error("Operation aborted");
|
|
}
|
|
|
|
const state = this.getOrCreateState(repositoryId);
|
|
|
|
if (!state.exclusiveHolder && state.sharedHolders.size === 0 && state.waitQueue.length === 0) {
|
|
const lockId = this.generateLockId();
|
|
state.exclusiveHolder = {
|
|
id: lockId,
|
|
operation,
|
|
acquiredAt: Date.now(),
|
|
};
|
|
return () => this.releaseExclusive(repositoryId, lockId);
|
|
}
|
|
|
|
logger.debug(
|
|
`[Mutex] Waiting for exclusive lock on repo ${repositoryId}: ${operation} (shared: ${state.sharedHolders.size}, exclusive: ${state.exclusiveHolder ? "yes" : "no"}, queue: ${state.waitQueue.length})`,
|
|
);
|
|
|
|
let onAbort: () => void = () => {};
|
|
let lockId: string | undefined;
|
|
try {
|
|
lockId = await new Promise<string>((resolve, reject) => {
|
|
const waiter = { type: "exclusive" as const, operation, resolve };
|
|
state.waitQueue.push(waiter);
|
|
|
|
if (signal) {
|
|
onAbort = () => {
|
|
const index = state.waitQueue.indexOf(waiter);
|
|
if (index !== -1) {
|
|
state.waitQueue.splice(index, 1);
|
|
this.cleanupStateIfEmpty(repositoryId);
|
|
this.notifyChange(repositoryId);
|
|
reject(signal.reason || new Error("Operation aborted"));
|
|
}
|
|
};
|
|
signal.addEventListener("abort", onAbort);
|
|
}
|
|
});
|
|
} finally {
|
|
signal?.removeEventListener("abort", onAbort);
|
|
}
|
|
|
|
if (signal?.aborted) {
|
|
if (lockId) {
|
|
this.releaseExclusive(repositoryId, lockId);
|
|
}
|
|
throw signal.reason || new Error("Operation aborted");
|
|
}
|
|
|
|
logger.debug(`[Mutex] Acquired exclusive lock for repo ${repositoryId}: ${operation} (${lockId})`);
|
|
|
|
return this.createRelease("exclusive", repositoryId, lockId!);
|
|
}
|
|
|
|
async acquireMany(requests: LockRequest[], signal?: AbortSignal) {
|
|
if (signal?.aborted) {
|
|
throw signal.reason || new Error("Operation aborted");
|
|
}
|
|
|
|
if (requests.length === 0) {
|
|
return () => {};
|
|
}
|
|
|
|
const seenRepositoryIds = new Set<string>();
|
|
for (const request of requests) {
|
|
if (seenRepositoryIds.has(request.repositoryId)) {
|
|
throw new Error(`Duplicate repository lock request: ${request.repositoryId}`);
|
|
}
|
|
seenRepositoryIds.add(request.repositoryId);
|
|
}
|
|
|
|
const sortedRequests = [...requests].sort((a, b) => a.repositoryId.localeCompare(b.repositoryId));
|
|
while (true) {
|
|
const releaseLocks = this.tryAcquireMany(sortedRequests);
|
|
if (releaseLocks) {
|
|
return releaseLocks;
|
|
}
|
|
|
|
await this.waitForChange(
|
|
sortedRequests.map((request) => request.repositoryId),
|
|
signal,
|
|
);
|
|
|
|
if (signal?.aborted) {
|
|
throw signal.reason || new Error("Operation aborted");
|
|
}
|
|
}
|
|
}
|
|
|
|
private releaseShared(repositoryId: string, lockId: string): void {
|
|
const state = this.locks.get(repositoryId);
|
|
if (!state) {
|
|
return;
|
|
}
|
|
|
|
const holder = state.sharedHolders.get(lockId);
|
|
if (!holder) {
|
|
return;
|
|
}
|
|
|
|
state.sharedHolders.delete(lockId);
|
|
const duration = Date.now() - holder.acquiredAt;
|
|
logger.debug(`[Mutex] Released shared lock for repo ${repositoryId}: ${holder.operation} (held for ${duration}ms)`);
|
|
|
|
this.processWaitQueue(repositoryId);
|
|
this.cleanupStateIfEmpty(repositoryId);
|
|
this.notifyChange(repositoryId);
|
|
}
|
|
|
|
private releaseExclusive(repositoryId: string, lockId: string): void {
|
|
const state = this.locks.get(repositoryId);
|
|
if (!state) {
|
|
return;
|
|
}
|
|
|
|
if (!state.exclusiveHolder || state.exclusiveHolder.id !== lockId) {
|
|
return;
|
|
}
|
|
|
|
const duration = Date.now() - state.exclusiveHolder.acquiredAt;
|
|
logger.debug(
|
|
`[Mutex] Released exclusive lock for repo ${repositoryId}: ${state.exclusiveHolder.operation} (held for ${duration}ms)`,
|
|
);
|
|
state.exclusiveHolder = null;
|
|
|
|
this.processWaitQueue(repositoryId);
|
|
this.cleanupStateIfEmpty(repositoryId);
|
|
this.notifyChange(repositoryId);
|
|
}
|
|
|
|
private processWaitQueue(repositoryId: string): void {
|
|
const state = this.locks.get(repositoryId);
|
|
if (!state || state.waitQueue.length === 0) {
|
|
return;
|
|
}
|
|
|
|
if (state.exclusiveHolder) {
|
|
return;
|
|
}
|
|
|
|
const firstWaiter = state.waitQueue[0];
|
|
|
|
if (firstWaiter.type === "exclusive") {
|
|
if (state.sharedHolders.size === 0) {
|
|
state.waitQueue.shift();
|
|
const lockId = this.generateLockId();
|
|
state.exclusiveHolder = {
|
|
id: lockId,
|
|
operation: firstWaiter.operation,
|
|
acquiredAt: Date.now(),
|
|
};
|
|
firstWaiter.resolve(lockId);
|
|
}
|
|
} else {
|
|
while (state.waitQueue.length > 0 && state.waitQueue[0].type === "shared") {
|
|
const waiter = state.waitQueue.shift();
|
|
if (!waiter) break;
|
|
const lockId = this.generateLockId();
|
|
state.sharedHolders.set(lockId, {
|
|
id: lockId,
|
|
operation: waiter.operation,
|
|
acquiredAt: Date.now(),
|
|
});
|
|
waiter.resolve(lockId);
|
|
}
|
|
}
|
|
}
|
|
|
|
isLocked(repositoryId: string): boolean {
|
|
const state = this.locks.get(repositoryId);
|
|
if (!state) return false;
|
|
return state.exclusiveHolder !== null || state.sharedHolders.size > 0;
|
|
}
|
|
|
|
private createRelease(type: LockType, repositoryId: string, lockId: string) {
|
|
let released = false;
|
|
return () => {
|
|
if (released) return;
|
|
released = true;
|
|
if (type === "shared") {
|
|
this.releaseShared(repositoryId, lockId);
|
|
} else {
|
|
this.releaseExclusive(repositoryId, lockId);
|
|
}
|
|
};
|
|
}
|
|
}
|
|
|
|
export const repoMutex = new RepositoryMutex();
|