More robust error handling.

Author: rkistner

packages/wa-sqlite-driver/src/pool.ts

@@ -11,12 +11,14 @@ import { WorkerDriverConnection } from './worker_threads';
 
 export function waSqliteWorkerPool(path: string): SqliteDriverConnectionPool {
   return new LazyConnectionPool(async () => {
-    return new WorkerDriverConnection(
+    const connection = new WorkerDriverConnection(
       new Worker(new URL('./wa-sqlite-worker.js', import.meta.url), {
         type: 'module'
       }),
       { path }
     );
+    await connection.open();
+    return connection;
     // return await WaSqliteConnection.open(path);
   });
 }

packages/wa-sqlite-driver/src/worker_threads/setup.ts

@@ -19,7 +19,7 @@ export function setupDriverWorker(config: WorkerDriverConfig) {
 
 export function setupDriverPort(config: WorkerDriverConfig) {
   let db: WorkerDriver | null = null;
-  let opened = new Deferred<void>();
+  let opened: Promise<void> | undefined = undefined;
   const port = {
     postMessage: self.postMessage.bind(self)
   };
@@ -30,47 +30,60 @@ export function setupDriverPort(config: WorkerDriverConfig) {
     console.log('received', message, id, args);
 
     if (message == 'open') {
+      const open = new Deferred<void>();
+      opened = open.promise;
       try {
         const connection = await config.openConnection(
           args as WorkerDriverConnectionOptions
         );
         db = new WorkerConnectionAdapter(connection);
         port.postMessage({ id });
-        opened.resolve();
+        open.resolve();
       } catch (e: any) {
-        opened.reject(e);
+        open.reject(e);
         port.postMessage({ id, value: { error: { message: e.message } } });
       }
     } else if (message == 'close') {
       try {
-        await opened.promise;
+        await opened;
         await db?.close();
         port.postMessage({ id });
       } catch (e: any) {
         port.postMessage({ id, value: { error: { message: e.message } } });
       }
     } else if (message == 'execute') {
-      await opened.promise;
-      const commands = args;
+      try {
+        await opened;
+        const commands = args;
 
-      const results = (await db!.execute(commands)).map((r) => {
-        if (isErrorResponse(r)) {
-          const error = r.error;
-          return {
+        const results = (await db!.execute(commands)).map((r) => {
+          if (isErrorResponse(r)) {
+            const error = r.error;
+            return {
+              error: {
+                code: error.code,
+                message: error.message,
+                stack: error.stack
+              }
+            };
+          } else {
+            return r;
+          }
+        });
+        port.postMessage({
+          id,
+          value: results
+        });
+      } catch (e) {
+        port.postMessage({
+          id,
+          value: args.map((c) => ({
             error: {
-              code: error.code,
-              message: error.message,
-              stack: error.stack
+              message: e.message
             }
-          };
-        } else {
-          return r;
-        }
-      });
-      port.postMessage({
-        id,
-        value: results
-      });
+          }))
+        });
+      }
     } else {
       throw new Error(`Unknown message: ${message}`);
     }

packages/wa-sqlite-driver/src/worker_threads/worker-driver.ts

@@ -38,11 +38,13 @@ export class WorkerDriverConnection implements SqliteDriverConnection {
   private ready: Promise<void>;
   private closing = false;
   private nextId = 1;
+  private options: WorkerDriverConnectionOptions;
 
   buffer: CommandQueueItem[] = [];
 
   constructor(worker: Worker, options: WorkerDriverConnectionOptions) {
     this.worker = worker;
+    this.options = options;
     worker.addEventListener('error', (err) => {
       console.error('worker error', err.message, err);
     });
@@ -62,10 +64,13 @@ export class WorkerDriverConnection implements SqliteDriverConnection {
         callback(value);
       });
     });
-    this.post('open', options);
     this.worker = worker;
   }
 
+  open() {
+    return this.post('open', this.options);
+  }
+
   prepare(sql: string, options?: PrepareOptions): WorkerDriverStatement {
     const id = this.nextId++;
     this.buffer.push({
@@ -125,7 +130,7 @@ export class WorkerDriverConnection implements SqliteDriverConnection {
         error: new SqliteError(error)
       } as any;
     }
-    return p;
+    return result;
   }
 
   async close() {
@@ -153,7 +158,9 @@ export class WorkerDriverConnection implements SqliteDriverConnection {
     for (let i = 0; i < commands.length; i++) {
       const c = commands[i];
       const rr = r[i];
-      if (isErrorResponse(rr)) {
+      if (rr == null) {
+        c.reject?.({ message: 'no result received', code: '' });
+      } else if (isErrorResponse(rr)) {
         c.reject?.(rr.error);
       } else if (c.resolve) {
         c.resolve!(rr.value);