Bugfix: fix race conditions

utils/sync-scheduler.ts

@@ -0,0 +1,63 @@
+type SyncTask = (dirPath: string, firstRun?: boolean) => Promise<void>;
+
+export default class SyncScheduler {
+    private pending = new Map<string, boolean>();
+    private timers = new Map<string, NodeJS.Timeout>();
+    private flushing = false;
+
+    constructor(
+        private readonly task: SyncTask,
+        private readonly debounceMs: number
+    ) {}
+
+    schedule(dirPath: string) {
+        const existing = this.timers.get(dirPath);
+        if (existing) clearTimeout(existing);
+
+        const timer = setTimeout(() => {
+            this.timers.delete(dirPath);
+            this.pending.set(dirPath, false);
+            void this.flush();
+        }, this.debounceMs);
+
+        this.timers.set(dirPath, timer);
+    }
+
+    enqueue(dirPath: string, firstRun?: boolean) {
+        const existing = this.timers.get(dirPath);
+        if (existing) {
+            clearTimeout(existing);
+            this.timers.delete(dirPath);
+        }
+
+        this.pending.set(dirPath, !!firstRun);
+        void this.flush();
+    }
+
+    private async flush() {
+        if (this.flushing) return;
+
+        this.flushing = true;
+
+        try {
+            while (this.pending.size > 0) {
+                const batch = [...this.pending.entries()];
+                this.pending.clear();
+
+                for (const [dirPath, firstRun] of batch) {
+                    try {
+                        await this.task(dirPath, firstRun);
+                    } catch (error: any) {
+                        console.log("ERROR:", error.message);
+                    }
+                }
+            }
+        } finally {
+            this.flushing = false;
+
+            if (this.pending.size > 0) {
+                void this.flush();
+            }
+        }
+    }
+}

utils/sync.ts

@@ -1,12 +1,68 @@
 import fs from "fs";
 import path from "path";
+import os from "os";
 import util from "util";
+import crypto from "crypto";
 import { exec } from "child_process";
 import { SyncFoldersSyncFnParams } from "../types";
 import grabDirNames from "./grab-dir-names";
+import { fldFileToStr } from "./grab-folders-files-string-paths";
+import delay from "./delay";
 
 const execPromise = util.promisify(exec);
 
+const LOCK_STALE_MS = 5 * 60 * 1000;
+const LOCK_RETRY_MS = 100;
+
+function lockPathFor(dirPath: string) {
+    const hash = crypto
+        .createHash("sha1")
+        .update(path.resolve(dirPath))
+        .digest("hex")
+        .slice(0, 16);
+
+    return path.join(os.tmpdir(), `turbosync-${hash}.lock`);
+}
+
+async function acquireLock(lockPath: string) {
+    while (true) {
+        try {
+            const fd = fs.openSync(lockPath, "wx");
+            fs.writeFileSync(fd, String(process.pid));
+            fs.closeSync(fd);
+            return;
+        } catch (error: any) {
+            if (error.code !== "EEXIST") throw error;
+
+            try {
+                const { mtimeMs } = fs.statSync(lockPath);
+                if (Date.now() - mtimeMs > LOCK_STALE_MS) {
+                    fs.unlinkSync(lockPath);
+                    continue;
+                }
+            } catch {
+                continue;
+            }
+
+            await delay(LOCK_RETRY_MS);
+        }
+    }
+}
+
+async function acquireLocks(lockPaths: string[]) {
+    for (const lockPath of lockPaths) {
+        await acquireLock(lockPath);
+    }
+}
+
+function releaseLocks(lockPaths: string[]) {
+    for (const lockPath of lockPaths) {
+        try {
+            fs.unlinkSync(lockPath);
+        } catch {}
+    }
+}
+
 export default async function sync({
     options,
     dirs,
@@ -94,11 +150,22 @@ export default async function sync({
         allCommandsArr.push(cmdArray);
     }
 
-    await Promise.all(
-        allCommandsArr.map((cmdArr) => {
-            return execPromise(cmdArr.join(" "));
-        })
-    );
+    const lockPaths = [dirPath, ...dstDirs.map((dr) => fldFileToStr(dr))]
+        .filter((pth): pth is string => Boolean(pth))
+        .map((pth) => lockPathFor(pth))
+        .sort();
+
+    await acquireLocks(lockPaths);
+
+    try {
+        await Promise.all(
+            allCommandsArr.map((cmdArr) => {
+                return execPromise(cmdArr.join(" "));
+            })
+        );
+    } finally {
+        releaseLocks(lockPaths);
+    }
 
     console.log(`${dirPath} Folder Sync Complete. Exiting ...`);
 }