chore: Trying to fix the pipeline

libs/as-sdk-integration-tests/src/http.test.ts

@@ -1,18 +1,19 @@
-import { callVm,  } from '../../../dist/libs/vm';
+import { callVm } from '../../../dist/libs/vm';
 import { jest } from '@jest/globals';
 import { readFile } from 'node:fs/promises';
 import { HttpFetchResponse } from '../../../dist/libs/vm/src/types/vm-actions';
 import { PromiseStatus } from '../../../dist/libs/vm/src/types/vm-promise';
 
 const mockHttpFetch = jest.fn();
 
+jest.setTimeout(30_000);
+
 const TestVmAdapter = jest.fn().mockImplementation(() => {
   return { httpFetch: mockHttpFetch };
 });
 
 describe('Http', () => {
   it('Test SDK HTTP Rejection', async () => {
-
     const wasmBinary = await readFile('dist/libs/as-sdk-integration-tests/debug.wasm');
     const result = await callVm({
       args: ['testHttpRejection'],

libs/vm/src/index.ts

@@ -5,6 +5,7 @@ import type { VmAdapter } from "./types/vm-adapter.js";
 import DefaultVmAdapter from "./default-vm-adapter.js";
 import { parse, format } from "node:path";
 import { HostToWorker } from "./worker-host-communication.js";
+import { createCallDataId } from "./utils.js";
 
 const CURRENT_FILE_PATH = parse(import.meta.url);
 CURRENT_FILE_PATH.base = 'worker.js';
@@ -25,18 +26,44 @@ export function callVm(callData: VmCallData, workerUrl = DEFAULT_WORKER_PATH, vm
       type: WorkerMessageType.VmCall,
     };
 
-    worker.on('message', (message: WorkerMessage) => {
-      if (message.type === WorkerMessageType.VmResult) {
-        resolve(message.result);
-      } else if (message.type === WorkerMessageType.VmActionExecute) {
-        hostToWorker.executeAction(message.action);
-      } else if (message.type === WorkerMessageType.VmActionResultBuffer) {
-        hostToWorker.sendActionResultToWorker(message.buffer);
-      } else {
-        console.warn('Unknown message', message);
+    setInterval(() => {
+      console.error("hey it has a worker", worker.threadId);
+    }, 2000);
+
+    worker.on('message', async (message: WorkerMessage) => {
+      try {
+        console.error('@message: ', message, createCallDataId(callData));
+        if (message.type === WorkerMessageType.VmResult) {
+          worker.terminate();
+          resolve(message.result);
+        } else if (message.type === WorkerMessageType.VmActionExecute) {
+          await hostToWorker.executeAction(message.action);
+        } else if (message.type === WorkerMessageType.VmActionResultBuffer) {
+          await hostToWorker.sendActionResultToWorker(message.buffer);
+        } else {
+          console.warn('Unknown message', message);
+        }
+      } catch (error) {
+        console.error('@callVm-onMessage: ', error, createCallDataId(callData));
       }
     });
 
+    worker.on('error', (error) => {
+      resolve({
+        exitCode: 1,
+        stderr: 'Worker threw an uncaught error: ' + error,
+        stdout: '',
+      });
+    });
+
+    worker.on('exit', (exitCode) => {
+      resolve({
+        exitCode,
+        stderr: 'The worker has been terminated',
+        stdout: '',
+      })
+    });
+
     worker.postMessage(workerMessage);
   });
 }

libs/vm/src/utils.ts

@@ -0,0 +1,11 @@
+import { createHash } from "node:crypto";
+import type { VmCallData } from "./vm";
+
+export function createCallDataId(callData: VmCallData): string {
+  const hasher = createHash('sha256');
+  const binary = new Uint8Array(callData.binary);
+  hasher.update(binary);
+  const binaryId = hasher.digest().toString('hex');
+
+  return `${callData.args.join('-')}-${binaryId}`;
+}

libs/vm/src/vm-imports.ts

@@ -16,14 +16,18 @@ export default class VmImports {
   }
 
   httpFetch(action: number, actionLength: number) {
+    console.log("HTTP Action called");
     const rawAction = new Uint8Array(
       this.memory?.buffer.slice(action, action + actionLength) ?? []
     );
     const messageRaw = Buffer.from(rawAction).toString('utf-8');
 
+    console.log('HTTP Action called with ', messageRaw);
+
     try {
       const message: HttpFetchAction = JSON.parse(messageRaw);
       this.callResult = this.workerToHost.callActionOnHost(message);
+      console.log('We got a result: ', this.callResult);
       return this.callResult.length;
     } catch (error) {
       console.error(`@httpFetch: ${messageRaw}`, error);

libs/vm/src/vm.test.ts

@@ -7,6 +7,8 @@ import { PromiseStatus } from "../../../dist/libs/vm/src/types/vm-promise";
 
 const mockHttpFetch = jest.fn();
 
+jest.setTimeout(10_000);
+
 const TestVmAdapter = jest.fn().mockImplementation(() => {
   return { httpFetch: mockHttpFetch };
 });

libs/vm/src/vm.ts

@@ -1,5 +1,6 @@
 import { init, WASI } from '@wasmer/wasi';
 import VmImports from './vm-imports.js';
+import { createCallDataId } from './utils.js';
 
 export interface VmCallData {
   binary: Uint8Array | number[];
@@ -34,7 +35,9 @@ export async function executeVm(callData: VmCallData, notifierBuffer: SharedArra
     const memory = instance.exports.memory;
     vmImports.setMemory(memory as WebAssembly.Memory);
 
+    console.log("Starting WASI", createCallDataId(callData));
     const exitCode = wasi.start(instance);
+    console.log("exit: ", exitCode, createCallDataId(callData));
 
     return {
       exitCode,
@@ -44,6 +47,7 @@ export async function executeVm(callData: VmCallData, notifierBuffer: SharedArra
       resultAsString: new TextDecoder().decode(vmImports.result),
     }
   } catch (err) {
+    console.log('Exception thrown', err, createCallDataId(callData));
     console.error(`
       @executeWasm
       Exception threw: ${err}
@@ -62,3 +66,4 @@ export async function executeVm(callData: VmCallData, notifierBuffer: SharedArra
     };
   }
 }
+

libs/vm/src/worker-host-communication.ts

@@ -69,6 +69,8 @@ export class HostToWorker {
     const notifierBufferi32 = new Int32Array(this.notifierBuffer);
     notifierBufferi32.set([this.actionResult.length], 1);
 
+    console.error("setted buffer, ready to send");
+
     updateNotifierState(notifierBufferi32, AtomicState.ResponseResultLength);
   }
 
@@ -88,7 +90,7 @@ export class HostToWorker {
 }
 
 export class WorkerToHost {
-  constructor(private notifierBuffer: SharedArrayBuffer) {}
+  constructor(private notifierBuffer: SharedArrayBuffer, private id: string = '') {}
 
   /**
    * Calls the given action on the host machine and sleeps the thread until an answer has been received
@@ -112,6 +114,8 @@ export class WorkerToHost {
 
     const length = notifierBufferi32[1];
 
+    console.error("The length is: ", length, this.id);
+
     // No need to do a full roundtrip if there are no bytes
     if (length === 0) {
       return Buffer.from([]);
@@ -123,6 +127,7 @@ export class WorkerToHost {
       type: WorkerMessageType.VmActionResultBuffer,
     };
 
+    console.error('Requesting message: ', message, this.id);
     parentPort?.postMessage(message);
 
     waitForNotifierStateChange(notifierBufferi32, AtomicState.RequestResult);

libs/vm/src/worker.ts

@@ -7,15 +7,19 @@ import {
 } from './types/worker-messages.js';
 
 parentPort?.on('message', async function (event) {
-  const message: WorkerMessage = event;
+  try {
+    const message: WorkerMessage = event;
 
-  if (message.type === WorkerMessageType.VmCall) {
-    const result = await executeVm(message.callData, message.notifierBuffer);
-    const response: VmResultWorkerMessage = {
-      result,
-      type: WorkerMessageType.VmResult,
-    };
+    if (message.type === WorkerMessageType.VmCall) {
+      const result = await executeVm(message.callData, message.notifierBuffer);
+      const response: VmResultWorkerMessage = {
+        result,
+        type: WorkerMessageType.VmResult,
+      };
 
-    parentPort?.postMessage(response);
+      parentPort?.postMessage(response);
+    }
+  } catch (error) {
+    console.error("@worker:message, error thrown: ", error);
   }
 });

package.json

@@ -52,6 +52,7 @@
     "dotenv": "^16.3.1",
     "figlet": "^1.6.0",
     "node-gzip": "^1.1.2",
+    "tslib": "^2.3.0",
     "ora": "^7.0.1",
     "protobufjs": "^7.2.5"
   }