fix(electron): should not continue pull when db closed (#2709)

(cherry picked from commit 008a05a470)

apps/electron/layers/main/src/db/__tests__/ensure-db.spec.ts

@@ -78,6 +78,10 @@ beforeEach(() => {
 
 afterEach(async () => {
   await runHandler('before-quit');
+  // wait for the db to be closed on Windows
+  if (process.platform === 'win32') {
+    await setTimeout(200);
+  }
   await fs.remove(tmpDir);
   vi.useRealTimers();
 });

apps/electron/layers/main/src/db/base-db-adapter.ts

@@ -18,6 +18,7 @@ export abstract class BaseSQLiteAdapter {
     if (!this.db) {
       this.db = new SqliteConnection(this.path);
       await this.db.connect();
+      logger.info(`[SQLiteAdapter:${this.role}]`, 'connected:', this.path);
     }
     return this.db;
   }

apps/electron/layers/main/src/db/secondary-db.ts

@@ -19,6 +19,7 @@ export class SecondaryWorkspaceSQLiteDB extends BaseSQLiteAdapter {
   role = 'secondary';
   yDoc = new Y.Doc();
   firstConnected = false;
+  destroyed = false;
 
   updateQueue: Uint8Array[] = [];
 
@@ -39,6 +40,7 @@ export class SecondaryWorkspaceSQLiteDB extends BaseSQLiteAdapter {
     this.unsubscribers.forEach(unsub => unsub());
     this.yDoc.destroy();
     await super.destroy();
+    this.destroyed = true;
   }
 
   get workspaceId() {
@@ -64,8 +66,9 @@ export class SecondaryWorkspaceSQLiteDB extends BaseSQLiteAdapter {
     );
     const updates = [...this.updateQueue];
     this.updateQueue = [];
-    await db.connect();
+    await this.run(async () => {
-    await this.addUpdateToSQLite(db, updates);
+      await this.addUpdateToSQLite(db, updates);
+    });
   }
 
   // flush after 5s, but will not wait for more than 10s
@@ -82,6 +85,9 @@ export class SecondaryWorkspaceSQLiteDB extends BaseSQLiteAdapter {
     (T extends (...args: any[]) => infer U ? Awaited<U> : unknown) | undefined
   > {
     try {
+      if (this.destroyed) {
+        return;
+      }
       await this.connectIfNeeded();
       this.runCounter++;
       return await fn();
@@ -90,6 +96,7 @@ export class SecondaryWorkspaceSQLiteDB extends BaseSQLiteAdapter {
     } finally {
       this.runCounter--;
       if (this.runCounter === 0) {
+        // just close db, but not the yDoc
         await super.destroy();
       }
     }
@@ -144,8 +151,11 @@ export class SecondaryWorkspaceSQLiteDB extends BaseSQLiteAdapter {
   // TODO: have a better solution to handle blobs
   async syncBlobs() {
     await this.run(async () => {
-      // pull blobs
+      // skip if upstream db is not connected (maybe it is already closed)
       const blobsKeys = await this.getBlobKeys();
+      if (!this.upstream.db || this.upstream.db?.isClose) {
+        return;
+      }
       const upstreamBlobsKeys = await this.upstream.getBlobKeys();
       // put every missing blob to upstream
       for (const key of blobsKeys) {
@@ -178,7 +188,7 @@ export class SecondaryWorkspaceSQLiteDB extends BaseSQLiteAdapter {
       return (await this.getUpdates()).map(update => update.data);
     });
 
-    if (!updates) {
+    if (!updates || this.destroyed) {
       return;
     }