fix(worker): isolate and wrapper lifetime fixes around startup and teardown

NativeScript/runtime/ConcurrentQueue.cpp

@@ -4,33 +4,29 @@
 namespace tns {
 
 void ConcurrentQueue::Initialize(CFRunLoopRef runLoop, void (*performWork)(void*), void* info) {
-    std::unique_lock<std::mutex> lock(initializationMutex_);
-    if (terminated) {
-        return;
-    }
+  std::unique_lock<std::mutex> lock(runLoopMutex_);
+  if (terminated) {
+    return;
+  }
     this->runLoop_ = runLoop;
     CFRunLoopSourceContext sourceContext = { 0, info, 0, 0, 0, 0, 0, 0, 0, performWork };
     this->runLoopTasksSource_ = CFRunLoopSourceCreate(kCFAllocatorDefault, 0, &sourceContext);
     CFRunLoopAddSource(this->runLoop_, this->runLoopTasksSource_, kCFRunLoopCommonModes);
 }
 
 void ConcurrentQueue::Push(std::shared_ptr<worker::Message> message) {
-    if (this->runLoopTasksSource_ != nullptr && !CFRunLoopSourceIsValid(this->runLoopTasksSource_)) {
-        return;
-    }
-
     {
-      // Checked under the queue mutex, where Terminate() also flips it while
-      // emptying the queue: a push that loses the race is dropped rather than
-      // landing in a queue nothing will ever pop again.
+      // Terminate() sets the flag before it empties the queue under this
+      // mutex, so a push either sees it here or lands ahead of that sweep:
+      // never in a queue nothing will pop again.
       std::unique_lock<std::mutex> mlock(this->mutex_);
       if (this->terminated) {
         return;
       }
         this->messagesQueue_.push(message);
     }
 
-    this->SignalAndWakeUp();
+    this->Signal();
 }
 
 std::vector<std::shared_ptr<worker::Message>> ConcurrentQueue::PopAll() {
@@ -52,20 +48,14 @@ bool ConcurrentQueue::IsEmpty() {
 }
 
 void ConcurrentQueue::Signal() {
-  // Mirrors Push()'s validity handling instead of SignalAndWakeUp()'s
-  // assert: a retry racing Terminate() must be a silent no-op.
-  if (this->runLoopTasksSource_ == nullptr ||
-      !CFRunLoopSourceIsValid(this->runLoopTasksSource_)) {
-    return;
+  // Serializes signaling and waking with Initialize() and Terminate().
+  // Terminate() clears both pointers and invalidates and releases the source;
+  // the run loop is borrowed from the worker thread, which terminates the
+  // queue before it leaves.
+  std::unique_lock<std::mutex> lock(runLoopMutex_);
+  if (this->runLoopTasksSource_ != nullptr) {
+    CFRunLoopSourceSignal(this->runLoopTasksSource_);
   }
-  this->SignalAndWakeUp();
-}
-
-void ConcurrentQueue::SignalAndWakeUp() {
-    if (this->runLoopTasksSource_ != nullptr) {
-        tns::Assert(CFRunLoopSourceIsValid(this->runLoopTasksSource_));
-        CFRunLoopSourceSignal(this->runLoopTasksSource_);
-    }
 
     if (this->runLoop_ != nullptr) {
         CFRunLoopWakeUp(this->runLoop_);
@@ -78,7 +68,7 @@ void ConcurrentQueue::Terminate() {
   // its sibling group's lock and posts to the sibling's loop.
   std::queue<std::shared_ptr<worker::Message>> dropped;
   {
-    std::unique_lock<std::mutex> lock(initializationMutex_);
+    std::unique_lock<std::mutex> lock(runLoopMutex_);
     terminated = true;
     CFRunLoopRef runLoop = this->runLoop_;
     CFRunLoopSourceRef source = this->runLoopTasksSource_;

NativeScript/runtime/ConcurrentQueue.h

@@ -31,8 +31,7 @@ struct ConcurrentQueue {
     CFRunLoopRef runLoop_ = nullptr;
     std::atomic<bool> terminated{false};
     std::mutex mutex_;
-    std::mutex initializationMutex_;
-    void SignalAndWakeUp();
+    std::mutex runLoopMutex_;
 };
 
 }

NativeScript/runtime/DataWrapper.h

@@ -553,12 +553,16 @@ class WorkerWrapper : public BaseDataWrapper {
   void Start(std::shared_ptr<v8::Persistent<v8::Value>> poWorker,
              std::function<v8::Isolate*()> func,
              std::optional<int> qualityOfService = std::nullopt);
-  void CallOnErrorHandlers(v8::TryCatch& tc);
+  // Both reporters take the isolate from their caller, which is running on
+  // it: they are reachable while the entry script is still evaluating, before
+  // workerIsolate_ is published.
+  void CallOnErrorHandlers(v8::Isolate* isolate, v8::TryCatch& tc);
   // Reports a rejected entry-evaluation promise. A rejection carries a reason
   // rather than a TryCatch, so it cannot go through CallOnErrorHandlers, but it
   // follows the same web order: the worker scope's `onerror` first, then — only
   // if that did not handle it — the parent's Worker error event.
-  void ReportEntryEvaluationRejection(v8::Local<v8::Context> context,
+  void ReportEntryEvaluationRejection(v8::Isolate* isolate,
+                                      v8::Local<v8::Context> context,
                                       v8::Local<v8::Value> reason);
   void PassUncaughtExceptionFromWorkerToMain(v8::Local<v8::Context> context,
                                              v8::TryCatch& tc,
@@ -622,7 +626,6 @@ class WorkerWrapper : public BaseDataWrapper {
   const inline v8::Isolate* GetMainIsolate() { return mainIsolate_; }
   // The only route from the worker thread to the parent: see mainLoop_.
   std::weak_ptr<EventLoop> MainLoop() const { return mainLoop_; }
-  const inline v8::Isolate* GetWorkerIsolate() { return workerIsolate_; }
 
   // Deletion is decided by who still holds the wrapper. The parent's Worker
   // object holds it from construction; Start() adds the worker thread. Each
@@ -647,7 +650,11 @@ class WorkerWrapper : public BaseDataWrapper {
   enum class Holders : uint8_t { Parent, Both, WorkerThread };
 
   v8::Isolate* mainIsolate_;
+  // Written by the worker thread only: published once the worker's startup
+  // function returns, withdrawn before the worker's runtime is deleted. Any
+  // other thread reads and uses it under workerIsolateMutex_.
   v8::Isolate* workerIsolate_;
+  std::mutex workerIsolateMutex_;
   std::atomic<bool> isRunning_;
   std::atomic<bool> isClosing_;
   std::atomic<bool> isTerminating_;

NativeScript/runtime/Worker.mm

@@ -564,7 +564,7 @@ throw NativeScriptException(
                                       ? info[0]
                                       : Local<Value>(v8::Exception::Error(tns::ToV8String(
                                             iso, "Worker entry module evaluation rejected")));
-            w->ReportEntryEvaluationRejection(ctx, reason);
+            w->ReportEntryEvaluationRejection(iso, ctx, reason);
           };
           Local<v8::Function> onFulfilled;
           Local<v8::Function> onRejected;
@@ -798,7 +798,7 @@ throw NativeScriptException(
     TryCatch tc(isolate);
     success = onCloseFunc->Call(context, v8::Undefined(isolate), 0, args).ToLocal(&result);
     if (!success && tc.HasCaught()) {
-      worker->CallOnErrorHandlers(tc);
+      worker->CallOnErrorHandlers(isolate, tc);
     }
   }
 }

NativeScript/runtime/WorkerWrapper.mm

@@ -122,17 +122,26 @@ static void PostToLoop(const std::shared_ptr<EventLoop>& loop, std::function<voi
 }
 
 void WorkerWrapper::EndWrapperLifetime() {
-  Local<Value> worker =
-      this->poWorker_ != nullptr ? this->poWorker_->Get(this->mainIsolate_) : Local<Value>();
+  // The dispatch below runs listeners, and a listener may shut the runtime
+  // down, whose teardown deletes this wrapper. Everything the dispatch needs is
+  // read first, and the liveness token says afterwards whether `this` is still
+  // there to unroot.
+  Isolate* isolate = this->mainIsolate_;
+  std::shared_ptr<std::atomic<WorkerWrapper*>> selfRef = this->selfRef_;
+  Local<Value> worker = this->poWorker_ != nullptr ? this->poWorker_->Get(isolate) : Local<Value>();
   if (!worker.IsEmpty() && worker->IsObject()) {
-    TryCatch tc(this->mainIsolate_);
-    Worker::EmitEnded(this->mainIsolate_, worker.As<Object>());
+    TryCatch tc(isolate);
+    Worker::EmitEnded(isolate, worker.As<Object>());
     if (tc.HasCaught()) {
       Local<Value> error = tc.Exception();
-      Log(@"%s", tns::ToString(this->mainIsolate_, error).c_str());
-      this->mainIsolate_->ThrowException(error);
+      Log(@"%s", tns::ToString(isolate, error).c_str());
+      isolate->ThrowException(error);
     }
   }
+  if (selfRef->load(std::memory_order_acquire) == nullptr) {
+    // Deleted during the dispatch; that teardown released the Worker object.
+    return;
+  }
   this->UnrootWorkerObject();
 }
 
@@ -183,7 +192,7 @@ static void PostToLoop(const std::shared_ptr<EventLoop>& loop, std::function<voi
     this->onMessage_(this->workerIsolate_, globalTarget, message);
 
     if (tc.HasCaught()) {
-      this->CallOnErrorHandlers(tc);
+      this->CallOnErrorHandlers(this->workerIsolate_, tc);
     }
   }
 
@@ -234,7 +243,11 @@ static void PostThreadEndedNotification(Isolate* mainIsolate, std::weak_ptr<Even
         },
         this);
 
-    this->workerIsolate_ = func();
+    Isolate* workerIsolate = func();
+    {
+      std::lock_guard<std::mutex> lock(this->workerIsolateMutex_);
+      this->workerIsolate_ = workerIsolate;
+    }
 
     this->DrainPendingTasks();
 
@@ -244,6 +257,20 @@ static void PostThreadEndedNotification(Isolate* mainIsolate, std::weak_ptr<Even
     }
   }
 
+  // The queue borrows this thread's run loop, so it lets go of it before the
+  // thread leaves: a terminate() that made this thread skip the loop above
+  // only reaches its own queue_.Terminate() later, possibly after the thread
+  // and its run loop are gone. A second Terminate() finds nothing set.
+  this->queue_.Terminate();
+
+  // Withdrawn before the runtime and its isolate go away below. Terminate()
+  // uses the isolate under this mutex, so a terminate that already read it has
+  // finished with it by the time this returns, and a later one finds null.
+  {
+    std::lock_guard<std::mutex> lock(this->workerIsolateMutex_);
+    this->workerIsolate_ = nullptr;
+  }
+
   // The inspector must be gone before the Runtime (and with it the isolate)
   // is deleted below.
   this->DestroyInspector();
@@ -315,20 +342,25 @@ static void PostThreadEndedNotification(Isolate* mainIsolate, std::weak_ptr<Even
   // set terminating to true atomically
   bool wasTerminating = this->isTerminating_.exchange(true);
   if (!wasTerminating) {
-    if (this->workerIsolate_ != nullptr) {
-      // Flagged before the request so a pump that is between iterations sees
-      // it on its next check, rather than only once V8 has some JS to
-      // interrupt — which a parked graph never provides.
-      //
-      // NOTE: `workerIsolate_` is assigned only after the worker's ENTRY has
-      // finished evaluating, so a worker still parked in its entry is not
-      // reachable from here at all and terminate() does nothing for it. That
-      // is a pre-existing worker-lifecycle gap, not something this flag can
-      // close — see the follow-up filed for it.
-      if (Runtime* workerRuntime = Runtime::GetRuntime(this->workerIsolate_)) {
-        workerRuntime->RequestTermination();
+    {
+      // Held across the use, not just the read: the worker thread withdraws
+      // the isolate under the same mutex before deleting its runtime.
+      std::lock_guard<std::mutex> lock(this->workerIsolateMutex_);
+      if (this->workerIsolate_ != nullptr) {
+        // Flagged before the request so a pump that is between iterations sees
+        // it on its next check, rather than only once V8 has some JS to
+        // interrupt — which a parked graph never provides.
+        //
+        // NOTE: `workerIsolate_` is assigned only after the worker's ENTRY has
+        // finished evaluating, so a worker still parked in its entry is not
+        // reachable from here at all and terminate() does nothing for it. That
+        // is a pre-existing worker-lifecycle gap, not something this flag can
+        // close — see the follow-up filed for it.
+        if (Runtime* workerRuntime = Runtime::GetRuntime(this->workerIsolate_)) {
+          workerRuntime->RequestTermination();
+        }
+        this->workerIsolate_->TerminateExecution();
       }
-      this->workerIsolate_->TerminateExecution();
     }
     {
       // A worker paused at a breakpoint sits in the inspector's nested pause
@@ -437,11 +469,10 @@ static void PostThreadEndedNotification(Isolate* mainIsolate, std::weak_ptr<Even
   delete client;
 }
 
-void WorkerWrapper::CallOnErrorHandlers(TryCatch& tc) {
+void WorkerWrapper::CallOnErrorHandlers(Isolate* isolate, TryCatch& tc) {
   if (this->isTerminating_) {
     return;
   }
-  Isolate* isolate = this->workerIsolate_;
   Local<Context> context = Caches::Get(isolate)->GetContext();
   Local<Object> global = context->Global();
 
@@ -470,11 +501,11 @@ static void PostThreadEndedNotification(Isolate* mainIsolate, std::weak_ptr<Even
   this->PassUncaughtExceptionFromWorkerToMain(context, tc);
 }
 
-void WorkerWrapper::ReportEntryEvaluationRejection(Local<Context> context, Local<Value> reason) {
+void WorkerWrapper::ReportEntryEvaluationRejection(Isolate* isolate, Local<Context> context,
+                                                   Local<Value> reason) {
   if (this->isTerminating_) {
     return;
   }
-  Isolate* isolate = this->workerIsolate_;
   Local<Object> global = context->Global();
 
   Local<Value> onErrorVal;

TestRunner/app/tests/MessagingTests.js

@@ -252,6 +252,16 @@ describe("Messaging runtime edges", function () {
                 }
             };
         });
+
+        it("reports an error onclose threw while the entry script was still running", function (done) {
+            var worker = new Worker("./messaging/throwingOncloseWorker.js");
+            worker.onerror = function (event) {
+                event.preventDefault();
+                expect(event.message).toContain("boom from onclose");
+                worker.terminate();
+                done();
+            };
+        });
     });
 
     describe("AbortSignal handler attribute accounting", function () {

TestRunner/app/tests/messaging/throwingOncloseWorker.js

@@ -0,0 +1,6 @@
+// Closes from inside the entry script, so onclose runs before the entry has
+// finished evaluating.
+onclose = function () {
+    throw new Error("boom from onclose");
+};
+close();