fix: Dont await on table resovler

Author: erezrokah

src/scheduler/scheduler.ts

@@ -92,20 +92,9 @@ const resolveTable = async (
 ) => {
   logger.info(`resolving table ${table.name}`);
   const stream = new TableResolverStream();
-  try {
-    await table.resolver(client, parent, stream);
-  } catch (error) {
-    const tableError = new SyncTableResolveError(`error resolving table ${table.name}`, {
-      cause: error,
-      props: { table, client },
-    });
-    logger.error(`error resolving table ${table.name}`, tableError);
-    return;
-  } finally {
-    stream.end();
-  }
+  const resolverPromise = table.resolver(client, parent, stream);
 
-  for await (const data of stream) {
+  const processData = async (data: unknown) => {
     logger.debug(`resolving resource for table ${table.name}`);
     const resolveResourceTimeout = 10 * 60 * 1000;
     const resource = new Resource(table, parent, data);
@@ -118,7 +107,7 @@ const resolveTable = async (
         props: { resource, table, client },
       });
       logger.error(preResolverError);
-      continue;
+      return;
     }
 
     try {
@@ -128,7 +117,7 @@ const resolveTable = async (
       await pTimeout(allColumnsPromise, { milliseconds: resolveResourceTimeout });
     } catch (error) {
       logger.error(`error resolving columns for table ${table.name}`, error);
-      continue;
+      return;
     }
 
     try {
@@ -139,7 +128,7 @@ const resolveTable = async (
         props: { resource, table, client },
       });
       logger.error(postResolveError);
-      continue;
+      return;
     }
 
     setCQId(resource, deterministicCQId);
@@ -148,7 +137,7 @@ const resolveTable = async (
       validateResource(resource);
     } catch (error) {
       logger.error(error);
-      continue;
+      return;
     }
 
     try {
@@ -161,14 +150,37 @@ const resolveTable = async (
         },
       });
       logger.error(encodeError);
-      continue;
+      return;
     }
 
     logger.debug(`done resolving resource for table ${table.name}`);
 
     await pMap(table.relations, (child) =>
       resolveTable(logger, client, child, resource, syncStream, deterministicCQId),
     );
+  };
+
+  const processPromises: Promise<void>[] = [];
+  stream.on('data', async (data) => {
+    const promise = processData(data);
+    processPromises.push(promise);
+  });
+
+  stream.on('end', async () => {
+    await Promise.all(processPromises);
+  });
+
+  try {
+    await resolverPromise;
+  } catch (error) {
+    const tableError = new SyncTableResolveError(`error resolving table ${table.name}`, {
+      cause: error,
+      props: { table, client },
+    });
+    logger.error(`error resolving table ${table.name}`, tableError);
+    return;
+  } finally {
+    stream.end();
   }
 
   logger.info(`done resolving table ${table.name}`);