[JS API] Improve error handling in AsyncInferQueue (#34857)

- exceptions thrown during asynchronous inference and callback execution
in `async_infer_queue.cpp` are now caught and converted to rejected
JavaScript promises
- test cases in `async_infer_queue.test.js` now use `try/finally` blocks
to guarantee that `inferQueue.release()` is called, even when tests fail
or throw exceptions.
- add tests to verify that `startAsync` correctly rejects promises when
provided with input data of incorrect shape or incompatible tensor types

### Tickets:
 - CVS-170804

Author: almilosz

src/bindings/js/node/src/async_infer_queue.cpp

@@ -108,22 +108,22 @@ void AsyncInferQueue::set_custom_callbacks(const Napi::CallbackInfo& info) {
         for (size_t handle = 0; handle < m_requests.size(); handle++) {
             m_requests[handle].set_callback([this, handle](std::exception_ptr exception_ptr) {
                 try {
-                    if (exception_ptr) {
-                        std::rethrow_exception(exception_ptr);
-                    }
-                    auto ov_callback = [this, handle](Napi::Env env, Napi::Function user_callback) {
+                    auto ov_callback = [this, handle, exception_ptr](Napi::Env env, Napi::Function user_callback) {
                         Napi::Object js_ir = InferRequestWrap::wrap(env, m_requests[handle]);
                         const auto promise = m_user_ids[handle].second;
                         try {
+                            if (exception_ptr) {
+                                std::rethrow_exception(exception_ptr);
+                            }
                             auto user_data =
                                 m_user_ids[handle].first.Value().ToString().Utf8Value() == UNDEFINED_USER_DATA
                                     ? env.Undefined()
                                     : m_user_ids[handle].first.Value();
-                            user_callback.Call({env.Null(), js_ir, user_data});  // CVS-170804
+                            user_callback.Call({env.Null(), js_ir, user_data});
                             promise.Resolve(user_data);
                             // returns before the promise's .then() is completed
-                        } catch (Napi::Error& e) {
-                            promise.Reject(Napi::Error::New(env, e.Message()).Value());
+                        } catch (const std::exception& e) {
+                            promise.Reject(Napi::Error::New(env, e.what()).Value());
                         }
                         // Start async inference on the next request or add idle handle to queue
                         if (std::lock_guard<std::mutex> lock(m_mutex); m_awaiting_requests.size() > 0) {
@@ -151,21 +151,27 @@ void AsyncInferQueue::start_async_impl(const size_t handle,
                                        Napi::Object infer_data,
                                        Napi::Object user_data,
                                        Napi::Promise::Deferred deferred) {
-    m_user_inputs[handle] = Napi::Persistent(infer_data);  // keep reference to inputs so they are not garbage collected
-    m_user_ids[handle] = std::make_pair(Napi::Persistent(user_data), deferred);
-
-    // CVS-166764
-    const auto& keys = infer_data.GetPropertyNames();
-    for (uint32_t i = 0; i < keys.Length(); ++i) {
-        auto input_name = static_cast<Napi::Value>(keys[i]).ToString().Utf8Value();
-        auto value = infer_data.Get(input_name);
-        auto tensor = value_to_tensor(value, m_requests[handle], input_name);
+    try {
+        m_user_inputs[handle] =
+            Napi::Persistent(infer_data);  // keep reference to inputs so they are not garbage collected
+        m_user_ids[handle] = std::make_pair(Napi::Persistent(user_data), deferred);
+
+        // CVS-166764
+        const auto& keys = infer_data.GetPropertyNames();
+        for (uint32_t i = 0; i < keys.Length(); ++i) {
+            auto input_name = static_cast<Napi::Value>(keys[i]).ToString().Utf8Value();
+            auto value = infer_data.Get(input_name);
+            auto tensor = value_to_tensor(value, m_requests[handle], input_name);
+
+            m_requests[handle].set_tensor(input_name, tensor);
+        }
 
-        m_requests[handle].set_tensor(input_name, tensor);
+        OPENVINO_ASSERT(m_tsfn != nullptr,
+                        "Callback has to be set before starting inference. Use 'setCallback' method.");
+        m_requests[handle].start_async();  // returns immediately, main event loop is free
+    } catch (const std::exception& e) {
+        deferred.Reject(Napi::Error::New(infer_data.Env(), e.what()).Value());
     }
-
-    OPENVINO_ASSERT(m_tsfn != nullptr, "Callback has to be set before starting inference. Use 'setCallback' method.");
-    m_requests[handle].start_async();  // returns immediately, main event loop is free
 }
 
 Napi::Value AsyncInferQueue::start_async(const Napi::CallbackInfo& info) {

src/bindings/js/node/tests/unit/async_infer_queue.test.js

@@ -60,15 +60,18 @@ describe("Tests for AsyncInferQueue.", () => {
 
     inferQueue.setCallback(callback);
 
-    for (let i = 0; i < jobs; i++) {
-      const img = generateImage();
-      // Start the inference request in non-blocking mode.
-      // The results will be available in the callback function.
-      await inferQueue.startAsync({ data: img }, i);
-    }
+    try {
+      for (let i = 0; i < jobs; i++) {
+        const img = generateImage();
+        // Start the inference request in non-blocking mode.
+        // The results will be available in the callback function.
+        await inferQueue.startAsync({ data: img }, i);
+      }
 
-    assert.strictEqual(jobsDone.filter((job) => job.finished).length, jobs);
-    inferQueue.release();
+      assert.strictEqual(jobsDone.filter((job) => job.finished).length, jobs);
+    } finally {
+      inferQueue.release();
+    }
   });
 
   it("Test startAsync fails without callback", async () => {
@@ -92,8 +95,11 @@ describe("Tests for AsyncInferQueue.", () => {
       }
     }
     inferQueue.setCallback(basicUserCallback);
-    await inferQueue.startAsync({ data: generateImage() });
-    inferQueue.release();
+    try {
+      await inferQueue.startAsync({ data: generateImage() });
+    } finally {
+      inferQueue.release();
+    }
   });
 
   it("test Promise.all()", async () => {
@@ -124,9 +130,17 @@ describe("Tests for AsyncInferQueue.", () => {
       promises.push(promise);
     }
 
-    await Promise.all(promises);
-    assert.strictEqual(jobsDone.filter((job) => job.finished).length, jobs);
-    inferQueue.release();
+    const allSettled = Promise.allSettled(promises);
+
+    try {
+      const results = await allSettled;
+      const rejected = results.filter((r) => r.status === "rejected");
+      assert.strictEqual(rejected.length, 0, "startAsync promise rejected");
+      assert.strictEqual(jobsDone.filter((job) => job.finished).length, jobs);
+    } finally {
+      await allSettled;
+      inferQueue.release();
+    }
   });
 
   it("Test AsyncInferQueue no freeze", async () => {
@@ -139,9 +153,12 @@ describe("Tests for AsyncInferQueue.", () => {
 
   it("Test double set_callback", async () => {
     const inferQueue = new ov.AsyncInferQueue(compiledModel, numRequest);
-    inferQueue.setCallback(() => {});
-    inferQueue.setCallback(() => {});
-    inferQueue.release();
+    try {
+      inferQueue.setCallback(() => {});
+      inferQueue.setCallback(() => {});
+    } finally {
+      inferQueue.release();
+    }
   });
 
   it("Test repeated AsyncInferQueue.release()", async () => {
@@ -165,19 +182,25 @@ describe("Tests for AsyncInferQueue.", () => {
 
   it("Test setCallback throws and list possible signatures", async () => {
     const inferQueue = new ov.AsyncInferQueue(compiledModel, numRequest);
-    assert.throws(() => {
-      inferQueue.setCallback();
-    }, /'setCallback' method called with incorrect parameters./);
-    inferQueue.release();
+    try {
+      assert.throws(() => {
+        inferQueue.setCallback();
+      }, /'setCallback' method called with incorrect parameters./);
+    } finally {
+      inferQueue.release();
+    }
   });
 
   it("Test startAsync throws and list possible signatures", async () => {
     const inferQueue = new ov.AsyncInferQueue(compiledModel, numRequest);
-    inferQueue.setCallback(basicUserCallback);
-    assert.throws(() => {
-      inferQueue.startAsync({ data: generateImage() }, "user_data", "extra_param");
-    }, /'startAsync' method called with incorrect parameters./);
-    inferQueue.release();
+    try {
+      inferQueue.setCallback(basicUserCallback);
+      assert.throws(() => {
+        inferQueue.startAsync({ data: generateImage() }, "user_data", "extra_param");
+      }, /'startAsync' method called with incorrect parameters./);
+    } finally {
+      inferQueue.release();
+    }
   });
 
   it("Test possibility to catch error in callback", async () => {
@@ -194,9 +217,12 @@ describe("Tests for AsyncInferQueue.", () => {
         }, /ReferenceError: jobsDone is not defined/);
       }
     }
-    inferQueue.setCallback(callback);
-    await inferQueue.startAsync({ data: generateImage() }, "data");
-    inferQueue.release();
+    try {
+      inferQueue.setCallback(callback);
+      await inferQueue.startAsync({ data: generateImage() }, "data");
+    } finally {
+      inferQueue.release();
+    }
   });
 
   it("Test error in callback and rejected promise", async () => {
@@ -212,11 +238,44 @@ describe("Tests for AsyncInferQueue.", () => {
       }
     }
 
-    inferQueue.setCallback(callback);
-    await inferQueue.startAsync({ data: generateImage() }, "data").catch((err) => {
-      assert(err instanceof Error);
-      assert.strictEqual(err.message, "jobsDone is not defined");
-    });
-    inferQueue.release();
+    try {
+      inferQueue.setCallback(callback);
+      await inferQueue.startAsync({ data: generateImage() }, "data").catch((err) => {
+        assert(err instanceof Error);
+        assert.strictEqual(err.message, "jobsDone is not defined");
+      });
+    } finally {
+      inferQueue.release();
+    }
+  });
+
+  it("Test startAsync with incorrect data shape", async () => {
+    const inferQueue = new ov.AsyncInferQueue(compiledModel, numRequest);
+    inferQueue.setCallback(basicUserCallback);
+
+    const incorrectShapeData = generateImage([1, 2, 3]);
+    try {
+      await assert.rejects(
+        async () => await inferQueue.startAsync({ data: incorrectShapeData }, "user_data"),
+        /Memory allocated using shape and element::type mismatc/,
+      );
+    } finally {
+      inferQueue.release();
+    }
+  });
+
+  it("Test startAsync with incorrect data shape2", async () => {
+    const inferQueue = new ov.AsyncInferQueue(compiledModel, numRequest);
+    inferQueue.setCallback(basicUserCallback);
+
+    const tensor1 = new ov.Tensor(ov.element.f32, [1, 3, 32, 31]);
+    try {
+      await assert.rejects(
+        async () => await inferQueue.startAsync({ data: tensor1 }, "user_data"),
+        /incompatible/,
+      );
+    } finally {
+      inferQueue.release();
+    }
   });
 });