raise limits on hot sync; make explicit remote scan much more efficient and scalable (in terms of number of files)

Author: williamstein

src/constants.ts

@@ -1,4 +1,4 @@
 export const CLI_NAME = "reflect-sync";
 export const SOFTWARE_NAME = "ReflectSync";
 
-export const MAX_WATCHERS = Number(process.env.REFLECT_MAX_WATCHERS ?? 64);
+export const MAX_WATCHERS = Number(process.env.REFLECT_MAX_WATCHERS ?? 128);

src/scan.ts

@@ -114,6 +114,11 @@ export function configureScanCommand(
       collectListOption,
       [] as string[],
     )
+    .option(
+      "--restricted-paths-stdin",
+      "read NUL-separated restricted paths from stdin",
+      false,
+    )
     .option("--numeric-ids", "include uid and gid in file hashes", false)
     .option(
       "-i, --ignore <pattern>",
@@ -167,6 +172,7 @@ type ScanOptions = {
   ignoreRules?: string[];
   ignore?: string[];
   restrictedPaths?: string[];
+  restrictedPathsStdin?: boolean;
   logicalClock?: LogicalClock;
   clockBase?: string[];
   scanTick?: number;
@@ -191,6 +197,7 @@ export async function runScan(opts: ScanOptions): Promise<void> {
     ignoreRules: ignoreRulesOpt = [],
     ignore: ignoreCliOpt = [],
     restrictedPaths = [],
+    restrictedPathsStdin = false,
     logicalClock,
     scanTick: scanTickOverride,
     filesystemCaps: providedCaps,
@@ -239,8 +246,13 @@ export async function runScan(opts: ScanOptions): Promise<void> {
   ignoreRaw.push(...autoIgnoreForRoot(absRoot, syncHome));
   const ignoreRules = normalizeIgnorePatterns(ignoreRaw);
 
+  const stdinRestricted: string[] = [];
+  if (restrictedPathsStdin) {
+    const buf = await readPathsFromStdin();
+    stdinRestricted.push(...buf);
+  }
   const restrictedPathList = dedupeRestrictedList(
-    includeAncestors(restrictedPaths),
+    includeAncestors([...restrictedPaths, ...stdinRestricted]),
   );
   const hasRestrictions = restrictedPathList.length > 0;
 
@@ -1536,6 +1548,20 @@ export async function runScan(opts: ScanOptions): Promise<void> {
   }
 }
 
+async function readPathsFromStdin(): Promise<string[]> {
+  return new Promise((resolve) => {
+    const chunks: Buffer[] = [];
+    process.stdin.on("data", (chunk) => chunks.push(chunk as Buffer));
+    process.stdin.on("end", () => {
+      if (chunks.length === 0) return resolve([]);
+      const buf = Buffer.concat(chunks);
+      const parts = buf.toString("utf8").split("\u0000");
+      resolve(parts.filter(Boolean));
+    });
+    process.stdin.resume();
+  });
+}
+
 // ---------- CLI entry (preserved) ----------
 cliEntrypoint<
   ScanOptions & {

src/scheduler.ts

@@ -94,7 +94,7 @@ const PRUNE_REMOTE_DATABASE_MS = 30_000;
 
 // never sync more than this many files at once using hot sync
 // (discards more)
-const MAX_HOT_SYNC = 200;
+const MAX_HOT_SYNC = 1_000;
 
 class FatalSchedulerError extends Error {
   constructor(message: string) {
@@ -137,6 +137,7 @@ export type SchedulerOptions = {
 
 type CycleOptions = {
   restrictedPaths?: string[];
+  restrictedPathsFromStdin?: boolean;
   label?: string;
 };
 
@@ -910,6 +911,7 @@ export async function runScheduler({
       numericIds?: boolean;
       ignoreRules: string[];
       restrictedPaths?: string[];
+      restrictedPathsStdin?: boolean;
       scanTick?: number;
       markCaseConflicts?: boolean;
       caseConflictCaps?: FilesystemCapabilities;
@@ -955,12 +957,18 @@ export async function runScheduler({
         sshArgs.push("--emit-since-age");
         sshArgs.push(`${Date.now() - lastRemoteScan.whenOk}`);
       }
-      if (params.restrictedPaths?.length) {
+      const stdinPaths = params.restrictedPathsStdin
+        ? (params.restrictedPaths ?? [])
+        : [];
+      if (params.restrictedPaths?.length && !params.restrictedPathsStdin) {
         for (const rel of params.restrictedPaths) {
           if (!rel) continue;
           sshArgs.push("--restricted-path", rel);
         }
       }
+      if (stdinPaths.length) {
+        sshArgs.push("--restricted-paths-stdin");
+      }
       if (params.markCaseConflicts) {
         sshArgs.push("--mark-case-conflicts");
         if (params.caseConflictCaps?.caseInsensitive) {
@@ -988,8 +996,13 @@ export async function runScheduler({
       );
 
       const sshP = spawn("ssh", sshArgs, {
-        stdio: ["ignore", "pipe", "pipe"],
+        stdio: params.restrictedPathsStdin
+          ? ["pipe", "pipe", "pipe"]
+          : ["ignore", "pipe", "pipe"],
       });
+      if (stdinPaths.length && sshP.stdin) {
+        sshP.stdin.end(stdinPaths.join("\u0000"));
+      }
       const stderrChunks: string[] = [];
       sshP.stderr?.on("data", (chunk) => {
         const text = chunk.toString();
@@ -1692,6 +1705,7 @@ export async function runScheduler({
       if (!expandedRestricted.length) return;
       await runCycle({
         restrictedPaths: expandedRestricted,
+        restrictedPathsFromStdin: alphaIsRemote || betaIsRemote,
         label: "hot",
       });
     } catch (e: any) {
@@ -1973,6 +1987,7 @@ export async function runScheduler({
             numericIds,
             ignoreRules,
             restrictedPaths,
+            restrictedPathsStdin: options.restrictedPathsFromStdin,
             scanTick,
             markCaseConflicts: markAlphaConflicts,
             caseConflictCaps: alphaCaseConflictCaps,
@@ -2051,6 +2066,8 @@ export async function runScheduler({
             numericIds,
             ignoreRules,
             restrictedPaths: hasRestrictions ? restrictedPaths : undefined,
+            restrictedPathsStdin:
+              options.restrictedPathsFromStdin && hasRestrictions,
             scanTick,
             markCaseConflicts: markBetaConflicts,
             caseConflictCaps: betaCaseConflictCaps,
@@ -2395,7 +2412,11 @@ export async function runScheduler({
             while (1) {
               let status;
               if (restrictedPaths?.length) {
-                status = await runCycle({ restrictedPaths, label: "hot" });
+                status = await runCycle({
+                  restrictedPaths,
+                  restrictedPathsFromStdin: alphaIsRemote || betaIsRemote,
+                  label: "hot",
+                });
               } else {
                 status = await runCycle();
               }