feat(async-worker): fix event-loop hang on env teardown

napi-inl.h

@@ -6746,6 +6746,25 @@ inline AsyncProgressWorkerBase<DataType>::AsyncProgressWorkerBase(
                                   /** context */ this,
                                   OnThreadSafeFunctionFinalize,
                                   /** finalizeData */ this);
+
+  // Register an environment cleanup hook to abort the thread-safe function and
+  // prevent event-loop hangs or running JavaScript if the environment
+  // terminates while the worker is active.
+  _cleanup_hook =
+      Env().AddCleanupHook(OnEnvironmentTeardown, static_cast<void*>(this));
+}
+
+template <typename DataType>
+inline void AsyncProgressWorkerBase<DataType>::OnEnvironmentTeardown(
+    void* arg) {
+  auto* self = static_cast<AsyncProgressWorkerBase<DataType>*>(arg);
+  if (self != nullptr) {
+    self->_teardown_called = true;
+    self->_cleanup_hook = BasicEnv::CleanupHook<void (*)(void*), void>();
+    if (self->_tsfn) {
+      self->_tsfn.Abort();
+    }
+  }
 }
 
 #if NAPI_VERSION > 4
@@ -6771,11 +6790,20 @@ inline AsyncProgressWorkerBase<DataType>::AsyncProgressWorkerBase(
                                   /** context */ this,
                                   OnThreadSafeFunctionFinalize,
                                   /** finalizeData */ this);
+
+  // Register an environment cleanup hook to abort the thread-safe function and
+  // prevent event-loop hangs or running JavaScript if the environment
+  // terminates while the worker is active.
+  _cleanup_hook =
+      env.AddCleanupHook(OnEnvironmentTeardown, static_cast<void*>(this));
 }
 #endif
 
 template <typename DataType>
 inline AsyncProgressWorkerBase<DataType>::~AsyncProgressWorkerBase() {
+  if (!_cleanup_hook.IsEmpty()) {
+    _cleanup_hook.Remove(Env());
+  }
   // Abort pending tsfn call.
   // Don't send progress events after we've already completed.
   // It's ok to call ThreadSafeFunction::Abort and ThreadSafeFunction::Release
@@ -6804,17 +6832,34 @@ inline napi_status AsyncProgressWorkerBase<DataType>::NonBlockingCall(
 
 template <typename DataType>
 inline void AsyncProgressWorkerBase<DataType>::OnWorkComplete(
-    Napi::Env /* env */, napi_status status) {
+    Napi::Env env, napi_status status) {
   _work_completed = true;
   _complete_status = status;
-  _tsfn.Release();
+  if (_tsfn_finalized || _teardown_called) {
+    AsyncWorker::OnWorkComplete(env, napi_cancelled);
+  } else {
+    _tsfn.Release();
+  }
 }
 
 template <typename DataType>
 inline void AsyncProgressWorkerBase<DataType>::OnThreadSafeFunctionFinalize(
     Napi::Env env, void* /* data */, AsyncProgressWorkerBase* context) {
+  context->_tsfn_finalized = true;
   if (context->_work_completed) {
-    context->AsyncWorker::OnWorkComplete(env, context->_complete_status);
+    napi_status status =
+        context->_teardown_called ? napi_cancelled : context->_complete_status;
+#ifdef NODE_ADDON_API_CPP_EXCEPTIONS
+    try {
+      context->AsyncWorker::OnWorkComplete(env, status);
+    } catch (const Error&) {
+      if (!context->_suppress_destruct) {
+        context->Destroy();
+      }
+    }
+#else
+    context->AsyncWorker::OnWorkComplete(env, status);
+#endif
   }
 }
 
@@ -7051,7 +7096,10 @@ inline void AsyncProgressQueueWorker<T>::SendProgress_(const T* data,
   std::copy(data, data + count, new_data);
 
   auto pair = new std::pair<T*, size_t>(new_data, count);
-  this->NonBlockingCall(pair);
+  if (this->NonBlockingCall(pair) != napi_ok) {
+    delete[] new_data;
+    delete pair;
+  }
 }
 
 template <class T>

napi.h

@@ -2863,7 +2863,9 @@ class AsyncWorker {
   ObjectReference _receiver;
   FunctionReference _callback;
   std::string _error;
-  bool _suppress_destruct;
+
+ protected:
+  bool _suppress_destruct = false;
 };
 #endif  // NAPI_HAS_THREADS
 
@@ -3338,10 +3340,14 @@ class AsyncProgressWorkerBase : public AsyncWorker {
 
  private:
   ThreadSafeFunction _tsfn;
+  BasicEnv::CleanupHook<void (*)(void*), void> _cleanup_hook;
   bool _work_completed = false;
+  bool _tsfn_finalized = false;
+  bool _teardown_called = false;
   napi_status _complete_status;
   static inline void OnThreadSafeFunctionFinalize(
       Napi::Env env, void* data, AsyncProgressWorkerBase* context);
+  static inline void OnEnvironmentTeardown(void* arg);
 };
 
 template <class T>

test/async_progress_worker.cc

@@ -2,6 +2,7 @@
 
 #include <chrono>
 #include <condition_variable>
+#include <future>
 #include <iostream>
 #include <mutex>
 #include <thread>
@@ -340,6 +341,65 @@ class SignalAfterProgressTestWorker : public AsyncProgressWorker<ProgressData> {
   }
   FunctionReference _progress;
 };
+
+std::promise<void> promise_for_worker_thread_;
+std::promise<void> promise_for_parent_thread_;
+
+class TerminatingWorker : public AsyncProgressWorker<ProgressData> {
+ public:
+  static void DoWork(const CallbackInfo& info) {
+    Function cb = info[0].As<Function>();
+    Function progress = info[1].As<Function>();
+
+    TerminatingWorker* worker = new TerminatingWorker(cb, progress);
+    worker->Queue();
+  }
+
+ protected:
+  void Execute(const ExecutionProgress& progress) override {
+    // Notify the parent thread that we are executing
+    promise_for_worker_thread_.set_value();
+
+    // Wait until parent thread initiates termination
+    std::future<void> future = promise_for_parent_thread_.get_future();
+    std::future_status status = future.wait_for(std::chrono::seconds(5));
+    if (status != std::future_status::ready) {
+      Error::Fatal("TerminatingWorker::Execute",
+                   "status != std::future_status::ready");
+    }
+
+    // Try to send progress during/after termination
+    ProgressData data{1};
+    progress.Send(&data, 1);
+  }
+
+  void OnProgress(const ProgressData* /* data */, size_t /* count */) override {
+  }
+
+ private:
+  TerminatingWorker(Function cb, Function progress) : AsyncProgressWorker(cb) {
+    _progress.Reset(progress, 1);
+  }
+  FunctionReference _progress;
+};
+
+void ResetPromises(const CallbackInfo&) {
+  promise_for_worker_thread_ = std::promise<void>();
+  promise_for_parent_thread_ = std::promise<void>();
+}
+
+void WaitForWorkerThread(const CallbackInfo&) {
+  std::future<void> future = promise_for_worker_thread_.get_future();
+  std::future_status status = future.wait_for(std::chrono::seconds(5));
+  if (status != std::future_status::ready) {
+    Error::Fatal("WaitForWorkerThread", "status != std::future_status::ready");
+  }
+}
+
+void ReleaseWorkerThread(const CallbackInfo&) {
+  promise_for_parent_thread_.set_value();
+}
+
 }  // namespace
 
 Object InitAsyncProgressWorker(Env env) {
@@ -351,6 +411,10 @@ Object InitAsyncProgressWorker(Env env) {
   exports["runWorkerNoCb"] = Function::New(env, TestWorkerWithNoCb::DoWork);
   exports["runWorkerWithRecv"] = Function::New(env, TestWorkerWithRecv::DoWork);
   exports["runWorkerWithCb"] = Function::New(env, TestWorkerWithCb::DoWork);
+  exports["resetPromises"] = Function::New(env, ResetPromises);
+  exports["waitForWorkerThread"] = Function::New(env, WaitForWorkerThread);
+  exports["releaseWorkerThread"] = Function::New(env, ReleaseWorkerThread);
+  exports["doTerminatingWork"] = Function::New(env, TerminatingWorker::DoWork);
   return exports;
 }
 

test/async_progress_worker_terminating_environment.js

@@ -0,0 +1,94 @@
+'use strict';
+
+const assert = require('assert');
+const { whichBuildType } = require('./common');
+
+// These tests ensure that AsyncProgressWorker can be used in a terminating
+// environment without triggering any crashes, use-after-free, or fatal errors.
+
+if (process.argv[2] === 'runInChildProcess') {
+  const bindingPath = process.argv[3];
+  const testCase = Number(process.argv[4]);
+
+  const binding = require(bindingPath);
+
+  binding.asyncprogressworker.resetPromises();
+
+  const { Worker } = require('worker_threads');
+
+  const worker = new Worker(
+    __filename,
+    {
+      argv: [
+        'runInWorkerThread',
+        bindingPath,
+        testCase
+      ]
+    }
+  );
+
+  if (testCase === 0) {
+    // Test case 0: Environment teardown occurs while worker is actively running
+    binding.asyncprogressworker.waitForWorkerThread();
+    const terminatePromise = worker.terminate();
+    binding.asyncprogressworker.releaseWorkerThread();
+    terminatePromise.then(() => {
+      process.exit(0);
+    });
+  } else if (testCase === 1) {
+    // Test case 1: Worker finishes normally, then environment is torn down
+    // (Verifies cleanup hook was safely removed and does not trigger use-after-free on teardown)
+    worker.on('exit', () => {
+      process.exit(0);
+    });
+  }
+} else if (process.argv[2] === 'runInWorkerThread') {
+  const bindingPath = process.argv[3];
+  const testCase = Number(process.argv[4]);
+
+  const binding = require(bindingPath);
+
+  if (testCase === 0) {
+    binding.asyncprogressworker.doTerminatingWork(
+      () => {},
+      () => {}
+    );
+  } else if (testCase === 1) {
+    binding.asyncprogressworker.doWork(
+      1,
+      () => {},
+      () => {}
+    );
+  }
+} else {
+  module.exports = wrapTest();
+
+  async function wrapTest () {
+    const path = require('path');
+    const buildType = await whichBuildType();
+    test(path.join(__dirname, `build/${buildType}/binding.node`));
+    test(path.join(__dirname, `build/${buildType}/binding_noexcept.node`));
+  }
+
+  function test (bindingPath) {
+    const numberOfTestCases = 2;
+
+    for (let i = 0; i < numberOfTestCases; ++i) {
+      const childProcess = require('./napi_child').spawnSync(
+        process.execPath,
+        [
+          __filename,
+          'runInChildProcess',
+          bindingPath,
+          i
+        ]
+      );
+
+      assert.strictEqual(
+        childProcess.status,
+        0,
+        `Test case ${bindingPath} ${i} failed with status ${childProcess.status}: ${childProcess.stderr}`
+      );
+    }
+  }
+}

test/index.js

@@ -109,6 +109,7 @@ if (napiVersion < 3) {
 if (napiVersion < 4 && !filterConditionsProvided) {
   testModules.splice(testModules.indexOf('asyncprogressqueueworker'), 1);
   testModules.splice(testModules.indexOf('asyncprogressworker'), 1);
+  testModules.splice(testModules.indexOf('async_progress_worker_terminating_environment'), 1);
   testModules.splice(testModules.indexOf('threadsafe_function/threadsafe_function_ctx'), 1);
   testModules.splice(testModules.indexOf('threadsafe_function/threadsafe_function_existing_tsfn'), 1);
   testModules.splice(testModules.indexOf('threadsafe_function/threadsafe_function_ptr'), 1);
@@ -131,6 +132,7 @@ if (napiVersion < 6 && !filterConditionsProvided) {
 if (majorNodeVersion < 12 && !filterConditionsProvided) {
   testModules.splice(testModules.indexOf('objectwrap_worker_thread'), 1);
   testModules.splice(testModules.indexOf('error_terminating_environment'), 1);
+  testModules.splice(testModules.indexOf('async_progress_worker_terminating_environment'), 1);
 }
 
 if (napiVersion < 8 && !filterConditionsProvided) {