additional error handling

Author: ptiurin

test/integration/v2/stream.test.ts

@@ -293,16 +293,26 @@ describe("streams", () => {
 
     // Add meta event handler to verify column metadata
     data.on("meta", m => {
-      expect(m).toEqual([
-        { name: "id", type: "int" },
-        { name: "username", type: "text" },
-        { name: "email", type: "text" },
-        { name: "status", type: "boolean" },
-        { name: "big_number", type: "long" },
-        { name: "created_date", type: "date" },
-        { name: "score", type: "double" },
-        { name: "description", type: "text" }
-      ]);
+      try {
+        expect(m).toEqual([
+          { name: "id", type: "int" },
+          { name: "username", type: "text" },
+          { name: "email", type: "text" },
+          { name: "status", type: "boolean" },
+          { name: "big_number", type: "long" },
+          { name: "created_date", type: "date" },
+          { name: "score", type: "double" },
+          { name: "description", type: "text" }
+        ]);
+      } catch (err) {
+        // Re-emit error so test fails properly
+        data.destroy(err as Error);
+      }
+    });
+
+    // Add error handler to ensure any stream errors are caught
+    data.on("error", err => {
+      throw err;
     });
 
     // Buffer pool configuration
@@ -398,15 +408,20 @@ describe("streams", () => {
     const outputStream = new stream.Transform({
       highWaterMark: 1,
       transform(chunk, encoding, callback) {
-        processedChunks++;
-
-        // Simulate occasional slow processing with minimal delays
-        if (processedChunks % 1000 === 0) {
-          setTimeout(() => {
+        try {
+          processedChunks++;
+
+          // Simulate occasional slow processing with minimal delays
+          if (processedChunks % 1000 === 0) {
+            // Use setImmediate instead of setTimeout to avoid potential timing issues
+            setImmediate(() => {
+              callback();
+            });
+          } else {
             callback();
-          }, 1); // 1ms delay
-        } else {
-          callback();
+          }
+        } catch (err) {
+          callback(err as Error);
         }
       }
     });