fix: eliminate JPromise listener leak and SafeContinuation deadlock (#1439)

example/src/getTests.ts

@@ -1787,6 +1787,36 @@ export function getTests(
           })
         ).didThrow()
     ),
+    ...('callCallbackSequentiallyFirstThrows' in testObject
+      ? [
+          createTest(
+            'Sequential async callback: second call succeeds when first threw',
+            async () =>
+              (
+                await it(async () => {
+                  let callCount = 0
+                  const result = await (
+                    testObject as TestObjectSwiftKotlin
+                  ).callCallbackSequentiallyFirstThrows(async (input) => {
+                    callCount++
+                    if (input.value === 'first') {
+                      throw new Error('Expected error on first call')
+                    }
+                    return input.value
+                  })
+                  if (callCount !== 2) {
+                    throw new Error(
+                      `Expected callback to be called twice, got ${callCount}`
+                    )
+                  }
+                  return result
+                })
+              )
+                .didNotThrow()
+                .equals('second')
+          ),
+        ]
+      : []),
     createTest('Getting complex callback from native returns a function', () =>
       it(() => testObject.getComplexCallback())
         .didNotThrow()

packages/react-native-nitro-modules/android/src/main/cpp/core/JPromise.hpp

@@ -74,15 +74,21 @@ class JPromise final : public jni::HybridClass<JPromise> {
     std::unique_lock lock(_mutex);
     jni::global_ref<jni::JObject> globalResult = jni::make_global(result);
     _state = globalResult;
-    for (const auto& onResolved : _onResolvedListeners) {
+    auto listeners = std::move(_onResolvedListeners);
+    _onRejectedListeners.clear(); // will never fire now
+    lock.unlock();
+    for (const auto& onResolved : listeners) {
       onResolved(result);
     }
   }
   void reject(jni::alias_ref<jni::JThrowable> error) {
     std::unique_lock lock(_mutex);
     jni::global_ref<jni::JThrowable> globalError = jni::make_global(error);
     _state = globalError;
-    for (const auto& onRejected : _onRejectedListeners) {
+    auto listeners = std::move(_onRejectedListeners);
+    _onResolvedListeners.clear(); // will never fire now
+    lock.unlock();
+    for (const auto& onRejected : listeners) {
       onRejected(error);
     }
   }
@@ -91,20 +97,24 @@ class JPromise final : public jni::HybridClass<JPromise> {
   void addOnResolvedListener(OnResolvedFunc&& onResolved) {
     std::unique_lock lock(_mutex);
     if (auto result = std::get_if<ResultType>(&_state)) {
-      // Promise is already resolved! Call the callback immediately
+      // Promise is already resolved — call immediately.
       onResolved(*result);
+    } else if (std::holds_alternative<ErrorType>(_state)) {
+      // Promise is already rejected — listener will never fire; discard it.
     } else {
-      // Promise is not yet resolved, put the listener in our queue.
+      // Promise is pending — queue the listener.
       _onResolvedListeners.push_back(std::move(onResolved));
     }
   }
   void addOnRejectedListener(OnRejectedFunc&& onRejected) {
     std::unique_lock lock(_mutex);
     if (auto error = std::get_if<ErrorType>(&_state)) {
-      // Promise is already rejected! Call the callback immediately
+      // Promise is already rejected — call immediately.
       onRejected(*error);
+    } else if (std::holds_alternative<ResultType>(_state)) {
+      // Promise is already resolved — listener will never fire; discard it.
     } else {
-      // Promise is not yet rejected, put the listener in our queue.
+      // Promise is pending — queue the listener.
       _onRejectedListeners.push_back(std::move(onRejected));
     }
   }
@@ -113,10 +123,12 @@ class JPromise final : public jni::HybridClass<JPromise> {
   void addOnResolvedListenerJava(jni::alias_ref<JOnResolvedCallback> callback) {
     std::unique_lock lock(_mutex);
     if (auto result = std::get_if<ResultType>(&_state)) {
-      // Promise is already resolved! Call the callback immediately
+      // Promise is already resolved — call immediately.
       callback->onResolved(*result);
+    } else if (std::holds_alternative<ErrorType>(_state)) {
+      // Promise is already rejected — listener will never fire; discard it.
     } else {
-      // Promise is not yet resolved, put the listener in our queue.
+      // Promise is pending — queue the listener.
       auto sharedCallback = jni::make_global(callback);
       _onResolvedListeners.emplace_back(
           [sharedCallback = std::move(sharedCallback)](const auto& result) { sharedCallback->onResolved(result); });
@@ -125,10 +137,12 @@ class JPromise final : public jni::HybridClass<JPromise> {
   void addOnRejectedListenerJava(jni::alias_ref<JOnRejectedCallback> callback) {
     std::unique_lock lock(_mutex);
     if (auto error = std::get_if<ErrorType>(&_state)) {
-      // Promise is already rejected! Call the callback immediately
+      // Promise is already rejected — call immediately.
       callback->onRejected(*error);
+    } else if (std::holds_alternative<ResultType>(_state)) {
+      // Promise is already resolved — listener will never fire; discard it.
     } else {
-      // Promise is not yet rejected, put the listener in our queue.
+      // Promise is pending — queue the listener.
       auto sharedCallback = jni::make_global(callback);
       _onRejectedListeners.emplace_back(
           [sharedCallback = std::move(sharedCallback)](const auto& error) { sharedCallback->onRejected(error); });

packages/react-native-nitro-modules/android/src/main/java/com/margelo/nitro/core/Promise.kt

@@ -7,9 +7,9 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
 import kotlin.concurrent.thread
-import kotlin.coroutines.resume
-import kotlin.coroutines.resumeWithException
-import kotlin.coroutines.suspendCoroutine
+import kotlin.coroutines.intrinsics.COROUTINE_SUSPENDED
+import kotlin.coroutines.intrinsics.intercepted
+import kotlin.coroutines.intrinsics.suspendCoroutineUninterceptedOrReturn
 
 /**
  * Represents a Promise that can be passed to JS.
@@ -86,12 +86,19 @@ class Promise<T> {
    * If the Promise is already resolved/rejected, this will continue immediately,
    * otherwise it will asynchronously wait for a result or throw on a rejection.
    * This function can only be used from a coroutine context.
+   *
+   * Uses [suspendCoroutineUninterceptedOrReturn] with [intercepted] to ensure
+   * [kotlin.coroutines.Continuation.resumeWith] is always called on the underlying
+   * continuation — including when called from Java with a raw [kotlin.coroutines.Continuation]
+   * (e.g. `CustomContinuation`) where [kotlin.coroutines.intrinsics.SafeContinuation]'s
+   * synchronous fast-path would otherwise skip [kotlin.coroutines.Continuation.resumeWith]
+   * entirely, leaving any `CompletableFuture.get()` blocked forever.
    */
-  suspend fun await(): T {
-    return suspendCoroutine { continuation ->
-      then { result -> continuation.resume(result) }
-      catch { error -> continuation.resumeWithException(error) }
-    }
+  suspend fun await(): T = suspendCoroutineUninterceptedOrReturn { uCont ->
+    val cont = uCont.intercepted()
+    then { result -> cont.resumeWith(Result.success(result)) }
+    catch { error -> cont.resumeWith(Result.failure(error)) }
+    COROUTINE_SUSPENDED
   }
 
   // C++ functions

packages/react-native-nitro-test/android/src/main/java/com/margelo/nitro/test/CustomContinuation.java

@@ -0,0 +1,59 @@
+package com.margelo.nitro.test;
+
+import org.jetbrains.annotations.NotNull;
+
+import java.util.concurrent.CompletableFuture;
+
+import kotlin.ResultKt;
+import kotlin.coroutines.Continuation;
+import kotlin.coroutines.CoroutineContext;
+import kotlin.coroutines.EmptyCoroutineContext;
+
+/**
+ * Bridges a Kotlin {@code Continuation<T>} to a Java {@link CompletableFuture}.
+ *
+ * <p>This is the <em>exact</em> pattern used in
+ * {@code JavaHybridAbtMobileClientSDK.java} (lines 627–641) that triggers
+ * https://github.com/mrousavy/nitro/issues/1439.</p>
+ *
+ * <p><strong>The bug:</strong> when the Kotlin {@code suspend fun await()} is called
+ * from Java with this continuation, and the {@code Promise} is <em>already resolved</em>
+ * by the time {@code await()} is called, Kotlin's {@code SafeContinuation} short-circuits:
+ * it returns the value directly from {@code await()} instead of calling
+ * {@link #resumeWith}. The Java caller ignores that return value, calls
+ * {@link CompletableFuture#get()}, and <strong>blocks forever</strong> because
+ * {@link #resumeWith} was never invoked.</p>
+ */
+public class CustomContinuation<T> implements Continuation<T> {
+
+    private final CompletableFuture<T> future;
+
+    public CustomContinuation(CompletableFuture<T> future) {
+        this.future = future;
+    }
+
+    @NotNull
+    @Override
+    public CoroutineContext getContext() {
+        return EmptyCoroutineContext.INSTANCE;
+    }
+
+    /**
+     * Called by Kotlin's coroutine machinery when the {@code Promise} resolves or
+     * rejects <em>asynchronously</em> (i.e. after {@code await()} returned
+     * {@code COROUTINE_SUSPENDED}).
+     *
+     * <p><em>Not</em> called when the Promise is already settled at the time
+     * {@code await()} is invoked — that is the root of the issue.</p>
+     */
+    @Override
+    public void resumeWith(@NotNull Object result) {
+        try {
+            ResultKt.throwOnFailure(result);
+            //noinspection unchecked
+            future.complete((T) result);
+        } catch (Throwable t) {
+            future.completeExceptionally(t);
+        }
+    }
+}

packages/react-native-nitro-test/android/src/main/java/com/margelo/nitro/test/HybridTestObjectKotlin.kt

@@ -14,6 +14,9 @@ import com.margelo.nitro.test.external.HybridSomeExternalObjectSpec
 import kotlinx.coroutines.delay
 import java.math.BigDecimal
 import java.time.Instant
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.TimeoutException
 
 @Keep
 @DoNotStrip
@@ -651,4 +654,38 @@ class HybridTestObjectKotlin : HybridTestObjectSwiftKotlinSpec() {
       .stripTrailingZeros()
       .toPlainString()
   }
+
+  // Single-threaded executor that reproduces the background-thread call pattern
+  // from real-world Java callers of Nitro (see issue #1439).
+  private val sequentialCallbackExecutor = Executors.newSingleThreadExecutor { r ->
+    Thread(r, "sequential-callback-thread").also { it.isDaemon = true }
+  }
+
+  override fun callCallbackSequentiallyFirstThrows(
+    fn: (input: SequentialCallbackInput) -> Promise<Promise<String>>,
+  ): Promise<String> {
+    val outerPromise = Promise<String>()
+    sequentialCallbackExecutor.submit {
+      try {
+        // First invocation — expected to throw.
+        try {
+          JavaCallHelper.awaitFnBlocking(fn(SequentialCallbackInput("first")), 3L, TimeUnit.SECONDS)
+          Log.w("HybridTestObjectKotlin", "First call unexpectedly succeeded")
+        } catch (_: TimeoutException) {
+          throw RuntimeException("callCallbackSequentiallyFirstThrows: first invocation timed out")
+        } catch (e: Exception) {
+          Log.d("HybridTestObjectKotlin", "First call threw as expected: ${e.message}")
+        }
+
+        // Second invocation — must not hang even though the inner Promise may already be resolved.
+        val result = JavaCallHelper.awaitFnBlocking(fn(SequentialCallbackInput("second")), 3L, TimeUnit.SECONDS)
+        outerPromise.resolve(result)
+      } catch (e: TimeoutException) {
+        outerPromise.reject(RuntimeException("callCallbackSequentiallyFirstThrows: second invocation timed out — SafeContinuation swallowed the resume", e))
+      } catch (e: Exception) {
+        outerPromise.reject(e)
+      }
+    }
+    return outerPromise
+  }
 }

packages/react-native-nitro-test/android/src/main/java/com/margelo/nitro/test/JavaCallHelper.java

@@ -0,0 +1,90 @@
+package com.margelo.nitro.test;
+
+import com.margelo.nitro.core.Promise;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ * Java helper that reproduces the exact JavaHybridAbtMobileClientSDK.java pattern
+ * (lines 627–641) which triggers https://github.com/mrousavy/nitro/issues/1439.
+ *
+ * <h2>Why this is in Java</h2>
+ * The bug requires calling the Kotlin {@code suspend fun Promise.await()} from Java
+ * with a custom {@link CustomContinuation}.  When Kotlin's {@code SafeContinuation}
+ * detects that the Promise is <em>already resolved</em>, it returns the result
+ * directly from {@code await()} — without ever calling
+ * {@link CustomContinuation#resumeWith}.  The Java caller ignores that return value
+ * and calls {@link CompletableFuture#get()}, which blocks forever.
+ *
+ * <h2>The pattern</h2>
+ * <pre>{@code
+ * Promise<Promise<String>> call = fn.invoke(input);
+ *
+ * // ─── outer promise ────────────────────────────────────────────────────────
+ * CompletableFuture<Promise<String>> future1 = new CompletableFuture<>();
+ * call.await(new CustomContinuation<>(future1));   // Java calls Kotlin suspend fun
+ * Promise<String> inner = future1.get(timeout);    // blocks until JS dispatches
+ *
+ * // ─── inner promise ────────────────────────────────────────────────────────
+ * CompletableFuture<String> future2 = new CompletableFuture<>();
+ * inner.await(new CustomContinuation<>(future2));  // ← BUG: if 'inner' is already
+ *                                                  //   resolved, SafeContinuation
+ *                                                  //   eats the value; future2
+ *                                                  //   is never completed
+ * return future2.get(timeout);                     // ← HANGS
+ * }</pre>
+ */
+public class JavaCallHelper {
+
+    /**
+     * Calls the given {@code call} Promise using the abt-SDK Java pattern.
+     *
+     * <p>This method deliberately reproduces the bug: if the inner Promise is
+     * already resolved by the time {@code inner.await(continuation)} is called,
+     * {@link CompletableFuture#get()} hangs forever (or until {@code timeout}
+     * expires with a {@link TimeoutException}).</p>
+     *
+     * @param call    the outer {@code Promise<Promise<String>>} returned by {@code fn.invoke(input)}
+     * @param timeout timeout value passed to each {@link CompletableFuture#get}
+     * @param unit    timeout unit
+     * @return the resolved string
+     * @throws TimeoutException    when the bug is triggered — inner future never completes
+     * @throws ExecutionException  when the JS function threw
+     * @throws InterruptedException if the waiting thread is interrupted
+     */
+    @SuppressWarnings("unused")
+    public static String awaitFnBlocking(
+            Promise<Promise<String>> call,
+            long timeout,
+            TimeUnit unit
+    ) throws Exception {
+
+        // ── Step 1: wait for the JS function to dispatch and return its Promise ──
+        CompletableFuture<Promise<String>> future1 = new CompletableFuture<>();
+        // Call the Kotlin `suspend fun await()` from Java with a CustomContinuation.
+        // When the outer Promise is pending this works correctly: COROUTINE_SUSPENDED is
+        // returned, and CustomContinuation.resumeWith() is called later on the JS thread.
+        call.await(new CustomContinuation<>(future1));
+        Promise<String> inner = unwrap(future1.get(timeout, unit));
+
+        // ── Step 2: wait for the actual string result ──────────────────────────
+        CompletableFuture<String> future2 = new CompletableFuture<>();
+        // BUG: if 'inner' is already resolved here (race between JS-thread resolution
+        // and the executor thread reaching this line), Kotlin's SafeContinuation
+        // returns "second" directly from await() without calling
+        // CustomContinuation.resumeWith().  future2 is never completed.
+        inner.await(new CustomContinuation<>(future2));
+        return unwrap(future2.get(timeout, unit)); // ← hangs when bug is triggered
+    }
+
+    /**
+     * Unwrap {@link ExecutionException} so callers see the original cause.
+     */
+    @SuppressWarnings("unchecked")
+    private static <T> T unwrap(T value) {
+        return value; // marker — real unwrapping is done by callers catching ExecutionException
+    }
+}

packages/react-native-nitro-test/ios/HybridTestObjectSwift.swift

@@ -625,4 +625,24 @@ class HybridTestObjectSwift: HybridTestObjectSwiftKotlinSpec {
   private func stringify(_ value: Double) -> String {
     return formatter.string(for: value) ?? "\(value)"
   }
+
+  func callCallbackSequentiallyFirstThrows(
+    fn: @escaping ((SequentialCallbackInput) -> Promise<Promise<String>>)
+  ) throws -> Promise<String> {
+    return Promise.async {
+      // First invocation — expected to throw.
+      do {
+        let innerPromise1 = try await fn(SequentialCallbackInput(value: "first")).await()
+        let result1 = try await innerPromise1.await()
+        print("[HybridTestObjectSwift] First call unexpectedly succeeded with: \(result1)")
+      } catch {
+        print("[HybridTestObjectSwift] First call threw as expected: \(error)")
+      }
+
+      // Second invocation — must not hang.
+      let innerPromise2 = try await fn(SequentialCallbackInput(value: "second")).await()
+      let result2 = try await innerPromise2.await()
+      return result2
+    }
+  }
 }