More sync engine tests + fixes

Author: zefhemel

client/codemirror/task.ts

@@ -1,5 +1,5 @@
 import { syntaxTree } from "@codemirror/language";
-import { startCompletion, completionStatus } from "@codemirror/autocomplete";
+import { startCompletion } from "@codemirror/autocomplete";
 import { Decoration, type EditorView, WidgetType } from "@codemirror/view";
 import type { NodeType } from "@lezer/common";
 import { decoratorStateField, isCursorInRange } from "./util.ts";

client/plugos/syscalls/sync.ts

@@ -2,6 +2,63 @@ import type { SysCallMapping } from "../system.ts";
 import type { Client } from "../../client.ts";
 
 export function syncSyscalls(client: Client): SysCallMapping {
+  const syncTimeoutMs = 30000;
+
+  function waitForServiceWorkerActivation(path?: string): Promise<any> {
+    return new Promise<any>((resolve, reject) => {
+      const timeout = setTimeout(() => {
+        cleanup();
+        reject(new Error(`Sync timeout after ${syncTimeoutMs / 1000}s`));
+      }, syncTimeoutMs);
+
+      function cleanup() {
+        clearTimeout(timeout);
+        client.eventHook.removeLocalListener(
+          "service-worker:file-sync-complete",
+          eventHandler,
+        );
+        client.eventHook.removeLocalListener(
+          "service-worker:space-sync-complete",
+          eventHandler,
+        );
+        client.eventHook.removeLocalListener(
+          "service-worker:sync-error",
+          errorHandler,
+        );
+      }
+
+      client.eventHook.addLocalListener(
+        "service-worker:file-sync-complete",
+        eventHandler,
+      );
+      client.eventHook.addLocalListener(
+        "service-worker:space-sync-complete",
+        eventHandler,
+      );
+      client.eventHook.addLocalListener(
+        "service-worker:sync-error",
+        errorHandler,
+      );
+
+      function eventHandler(data: any) {
+        if (data.path && path && data.path !== path) {
+          return;
+        }
+        cleanup();
+        resolve(data);
+      }
+
+      function errorHandler(e: any) {
+        // Only reject if the error is for our specific path, or if no path was specified
+        if (e.path && path && e.path !== path) {
+          return;
+        }
+        cleanup();
+        reject(e);
+      }
+    });
+  }
+
   return {
     "sync.hasInitialSyncCompleted": (): boolean => {
       return client.fullSyncCompleted;
@@ -14,61 +71,16 @@ export function syncSyscalls(client: Client): SysCallMapping {
       // postServiceWorkerMessage returns silently if no SW, so only wait if SW is active
       const registration = await navigator.serviceWorker.getRegistration();
       if (registration?.active) {
-        return waitForServiceWorkerActivation(client, path);
+        return waitForServiceWorkerActivation(path);
       }
     },
     "sync.performSpaceSync": async (): Promise<number> => {
       await client.postServiceWorkerMessage({ type: "perform-space-sync" });
       const registration = await navigator.serviceWorker.getRegistration();
       if (registration?.active) {
-        return waitForServiceWorkerActivation(client);
+        return waitForServiceWorkerActivation();
       }
       return 0;
     },
   };
 }
-
-function waitForServiceWorkerActivation(
-  client: Client,
-  path?: string,
-): Promise<any> {
-  return new Promise<any>((resolve, reject) => {
-    client.eventHook.addLocalListener(
-      "service-worker:file-sync-complete",
-      eventHandler,
-    );
-    client.eventHook.addLocalListener(
-      "service-worker:space-sync-complete",
-      eventHandler,
-    );
-    client.eventHook.addLocalListener(
-      "service-worker:sync-error",
-      errorHandler,
-    );
-    function eventHandler(data: any) {
-      if (data.path && path && data.path !== path) {
-        return;
-      }
-      resolve(data);
-      cleanup();
-    }
-    function errorHandler(e: any) {
-      reject(e);
-      cleanup();
-    }
-    function cleanup() {
-      client.eventHook.removeLocalListener(
-        "service-worker:file-sync-complete",
-        eventHandler,
-      );
-      client.eventHook.removeLocalListener(
-        "service-worker:space-sync-complete",
-        eventHandler,
-      );
-      client.eventHook.removeLocalListener(
-        "service-worker:sync-error",
-        errorHandler,
-      );
-    }
-  });
-}

client/service_worker.ts

@@ -288,10 +288,11 @@ self.addEventListener("message", async (event: any) => {
               operations,
             });
           },
-          syncError: (error) => {
+          syncError: (error, path) => {
             broadcastMessage({
               type: "sync-error",
               message: error.message,
+              path,
             });
           },
         });

client/service_worker/proxy_router.ts

@@ -233,11 +233,9 @@ export class ProxyRouter extends EventEmitter<ProxyRouterEvents> {
 
     const files = await this.localSpacePrimitives.fetchFileList();
     // Now augment this with non-synced file metadata
+    const localFileNames = new Set(files.map((f) => f.name));
     for (const nonSyncedFile of this.nonSyncedFiles.values()) {
-      const existingFile = files.find(
-        (file) => file.name === nonSyncedFile.name,
-      );
-      if (!existingFile) {
+      if (!localFileNames.has(nonSyncedFile.name)) {
         files.push(nonSyncedFile);
       }
     }

client/service_worker/sync_engine.ts

@@ -17,7 +17,7 @@ type SyncEngineEvents = {
   // A single file syncle has completed
   fileSyncComplete: (path: string, operations: number) => void | Promise<void>;
 
-  syncError: (error: Error) => void | Promise<void>;
+  syncError: (error: Error, path?: string) => void | Promise<void>;
 
   // Sync conflict occurred
   syncConflict: (path: string) => void | Promise<void>;
@@ -139,7 +139,7 @@ export class SyncEngine extends EventEmitter<SyncEngineEvents> {
       void this.emit("fileSyncComplete", path, operations);
       return operations;
     } catch (e) {
-      void this.emit("syncError", e);
+      void this.emit("syncError", e, path);
       throw e;
     }
   }

client/spaces/checked_space_primitives.ts

@@ -21,14 +21,14 @@ export class CheckedSpacePrimitives implements SpacePrimitives {
 
   readFile(path: string): Promise<{ data: Uint8Array; meta: FileMeta }> {
     if (!this.isReadable(path)) {
-      throw new Error("Couldn't write file, path isn't writable");
+      throw new Error("Couldn't read file, path isn't readable");
     }
     return this.wrapped.readFile(path);
   }
 
   getFileMeta(path: string, observing?: boolean): Promise<FileMeta> {
     if (!this.isReadable(path)) {
-      throw new Error("Couldn't get file meta, path isn't writable");
+      throw new Error("Couldn't get file meta, path isn't readable");
     }
     return this.wrapped.getFileMeta(path, observing);
   }

client/spaces/evented_space_primitives.ts

@@ -14,10 +14,13 @@ import type { DataStore } from "../data/datastore.ts";
  * - page:deleted (string)
  */
 export class EventedSpacePrimitives implements SpacePrimitives {
-  // Various operations may be going on at the same time, and we don't want to trigger events unnessarily.
-  // Therefore, we use this variable to track if any operation is in flight, and if so, we skip event triggering.
-  // This is ok, because any event will be picked up in a following iteration.
-  operationInProgress = false;
+  // Various operations may be going on at the same time, and we don't want to trigger events unnecessarily.
+  // Therefore, we use this counter to track how many operations are in flight, and if so, we skip event triggering.
+  private operationCount = 0;
+
+  // When a fetchFileList is requested while operations are in flight, we defer it
+  // so that synced changes are not missed.
+  private deferredFetchFileList = false;
 
   private enabled = false;
 
@@ -65,6 +68,20 @@ export class EventedSpacePrimitives implements SpacePrimitives {
     }
   }
 
+  /**
+   * Called when an operation completes. If a fetchFileList was deferred
+   * because operations were in flight, trigger it now.
+   */
+  private checkDeferredFetchFileList() {
+    if (this.deferredFetchFileList && this.operationCount === 0) {
+      this.deferredFetchFileList = false;
+      // Schedule on next tick to avoid reentrancy
+      setTimeout(() => {
+        void this.fetchFileList();
+      });
+    }
+  }
+
   dispatchEvent(name: string, ...args: any[]): Promise<any[]> {
     if (!this.enabled) {
       return Promise.resolve([]);
@@ -74,27 +91,22 @@ export class EventedSpacePrimitives implements SpacePrimitives {
   }
 
   async fetchFileList(): Promise<FileMeta[]> {
-    if (this.operationInProgress) {
+    if (this.operationCount > 0) {
       // Some other operation (read, write, list, meta) is already going on
       // this will likely trigger events, so let's not worry about any of that and avoid race condition and inconsistent data.
+      // We mark a deferred flag so the next operation completion will trigger a fetchFileList.
       console.info(
-        "operationInProgress, deferring event processing for fetchFileList.",
+        "deferredFetchFileList: skipping event triggering for fetchFileList.",
       );
-      const result = await this.wrapped.fetchFileList();
-      // Schedule a retry after current operation completes
-      setTimeout(() => {
-        if (!this.operationInProgress) {
-          void this.fetchFileList();
-        }
-      }, 50);
-      return result;
+      this.deferredFetchFileList = true;
+      return this.wrapped.fetchFileList();
     }
     if (!this.enabled) {
       return this.wrapped.fetchFileList();
     }
     // console.log("Fetching file list");
     // Fetching mutex
-    this.operationInProgress = true;
+    this.operationCount++;
     try {
       // Fetch the list
       const newFileList = await this.wrapped.fetchFileList();
@@ -140,27 +152,25 @@ export class EventedSpacePrimitives implements SpacePrimitives {
       return newFileList;
     } finally {
       await this.saveSnapshot();
-      this.operationInProgress = false;
+      this.operationCount--;
     }
   }
 
   async readFile(path: string): Promise<{ data: Uint8Array; meta: FileMeta }> {
     if (!this.enabled) {
       return this.wrapped.readFile(path);
     }
+    this.operationCount++;
     try {
-      // Fetching mutex
-      const wasFetching = this.operationInProgress;
-      this.operationInProgress = true;
-
       // Fetch file
       const data = await this.wrapped.readFile(path);
-      if (!wasFetching) {
+      if (this.operationCount === 1) {
         await this.triggerEventsAndCache(path, data.meta.lastModified);
       }
       return data;
     } finally {
-      this.operationInProgress = false;
+      this.operationCount--;
+      this.checkDeferredFetchFileList();
     }
   }
 
@@ -173,11 +183,10 @@ export class EventedSpacePrimitives implements SpacePrimitives {
       return this.wrapped.writeFile(path, data, meta);
     }
 
-    const wasFetching = this.operationInProgress;
-    this.operationInProgress = true;
+    this.operationCount++;
     try {
       const newMeta = await this.wrapped.writeFile(path, data, meta);
-      if (!wasFetching) {
+      if (this.operationCount === 1) {
         await this.triggerEventsAndCache(path, newMeta.lastModified);
       }
       if (path.endsWith(".md")) {
@@ -187,7 +196,8 @@ export class EventedSpacePrimitives implements SpacePrimitives {
 
       return newMeta;
     } finally {
-      this.operationInProgress = false;
+      this.operationCount--;
+      this.checkDeferredFetchFileList();
     }
   }
 
@@ -212,16 +222,16 @@ export class EventedSpacePrimitives implements SpacePrimitives {
       return this.wrapped.getFileMeta(path, observing);
     }
 
+    this.operationCount++;
     try {
-      const wasFetching = this.operationInProgress;
-      this.operationInProgress = true;
       const newMeta = await this.wrapped.getFileMeta(path, observing);
-      if (!wasFetching) {
+      if (this.operationCount === 1) {
         await this.triggerEventsAndCache(path, newMeta.lastModified);
       }
       return newMeta;
     } finally {
-      this.operationInProgress = false;
+      this.operationCount--;
+      this.checkDeferredFetchFileList();
     }
   }
 
@@ -230,19 +240,19 @@ export class EventedSpacePrimitives implements SpacePrimitives {
       return this.wrapped.deleteFile(path);
     }
 
+    this.operationCount++;
     try {
-      this.operationInProgress = true;
       if (path.endsWith(".md")) {
         const pageName = path.substring(0, path.length - 3);
         await this.dispatchEvent("page:deleted", pageName);
       }
-      // await this.getPageMeta(path); // Check if page exists, if not throws Error
       await this.wrapped.deleteFile(path);
       this.deleteFromSnapshot(path);
       await this.dispatchEvent("file:deleted", path);
     } finally {
       await this.saveSnapshot();
-      this.operationInProgress = false;
+      this.operationCount--;
+      this.checkDeferredFetchFileList();
     }
   }
 }