[CDX-358] Handle undeliverable RxJava exceptions

library/src/main/java/io/constructor/core/ConstructorIo.kt

@@ -33,7 +33,10 @@ import io.constructor.util.urlEncode
 import io.reactivex.Completable
 import io.reactivex.Observable
 import io.reactivex.disposables.CompositeDisposable
+import io.reactivex.exceptions.UndeliverableException
+import io.reactivex.plugins.RxJavaPlugins
 import io.reactivex.schedulers.Schedulers
+import java.io.IOException
 import java.util.*
 
 typealias ConstructorError = ((Throwable) -> Unit)?
@@ -110,6 +113,10 @@ object ConstructorIo {
         }
         this.context = context.applicationContext
 
+        // Install the error handler early, before component initialization triggers the Dagger
+        // graph, so that any undeliverable RxJava exceptions during init are also caught.
+        setupRxJavaErrorHandler()
+
         configMemoryHolder = component.configMemoryHolder()
         configMemoryHolder.autocompleteResultCount = constructorIoConfig.autocompleteResultCount
         configMemoryHolder.testCellParams = constructorIoConfig.testCells
@@ -131,6 +138,49 @@ object ConstructorIo {
         dataManager = component.dataManager()
     }
 
+    /**
+     * Sets up a global RxJava error handler to gracefully handle undeliverable exceptions.
+     * These exceptions can occur when network errors happen after the RxJava stream has
+     * already completed or been disposed, particularly with OkHttp async operations.
+     */
+    @Synchronized
+    internal fun setupRxJavaErrorHandler() {
+        if (RxJavaPlugins.getErrorHandler() != null) return
+        RxJavaPlugins.setErrorHandler { throwable ->
+            var error = throwable
+            // Unwrap the actual cause from UndeliverableException
+            if (error is UndeliverableException) {
+                error = error.cause ?: error
+            }
+
+            // InterruptedException signals the thread should stop — restore the interrupt flag
+            // so cooperative cancellation continues to work, then return without crashing
+            if (error is InterruptedException) {
+                Thread.currentThread().interrupt()
+                e("Constructor.io: Non-fatal interrupted error: ${error.message}")
+                return@setErrorHandler
+            }
+
+            // Network exceptions are expected during normal operation (timeout, no connectivity, etc.)
+            // Log them but don't crash the app
+            if (error is IOException) {
+                e("Constructor.io: Non-fatal network error: ${error.javaClass.simpleName} - ${error.message}")
+                return@setErrorHandler
+            }
+
+            // Unexpected exception — forward to the thread's uncaught exception handler,
+            // falling back to the JVM default handler if the thread has none set.
+            // If no handler is available, log the error.
+            val handler = Thread.currentThread().uncaughtExceptionHandler
+                ?: Thread.getDefaultUncaughtExceptionHandler()
+            if (handler != null) {
+                handler.uncaughtException(Thread.currentThread(), error)
+            } else {
+                e("Constructor.io: Undeliverable unexpected exception (no uncaught handler): $error")
+            }
+        }
+    }
+
     /**
      * Returns the current session identifier (an incrementing integer)
      */

library/src/main/java/io/constructor/injection/module/NetworkModule.kt

@@ -9,6 +9,7 @@ import io.constructor.data.local.PreferencesHelper
 import io.constructor.data.memory.ConfigMemoryHolder
 import io.constructor.data.model.dataadapter.ResultDataAdapter
 import io.constructor.injection.ConstructorSdk
+import io.reactivex.schedulers.Schedulers
 import okhttp3.OkHttpClient
 import okhttp3.logging.HttpLoggingInterceptor
 import retrofit2.Retrofit
@@ -33,7 +34,7 @@ object NetworkModule {
             .baseUrl(preferencesHelper.scheme + "://" + preferencesHelper.serviceUrl)
             .client(okHttpClient)
             .addConverterFactory(MoshiConverterFactory.create(moshi))
-            .addCallAdapterFactory(RxJava2CallAdapterFactory.create())
+            .addCallAdapterFactory(RxJava2CallAdapterFactory.createWithScheduler(Schedulers.io()))
             .build()
 
     @Provides

library/src/test/java/io/constructor/core/ConstructorIoRxErrorHandlerTest.kt

@@ -0,0 +1,122 @@
+package io.constructor.core
+
+import io.reactivex.exceptions.UndeliverableException
+import io.reactivex.functions.Consumer
+import io.reactivex.plugins.RxJavaPlugins
+import org.junit.After
+import org.junit.Assert.assertSame
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import java.io.IOException
+import java.net.SocketTimeoutException
+
+class ConstructorIoRxErrorHandlerTest {
+
+    @Before
+    fun setup() {
+        RxJavaPlugins.reset()
+        ConstructorIo.setupRxJavaErrorHandler()
+    }
+
+    @After
+    fun teardown() {
+        RxJavaPlugins.reset()
+    }
+
+    @Test
+    fun isIdempotent() {
+        val handlerAfterFirstCall = RxJavaPlugins.getErrorHandler()
+        ConstructorIo.setupRxJavaErrorHandler()
+        assertSame(handlerAfterFirstCall, RxJavaPlugins.getErrorHandler())
+    }
+
+    @Test
+    fun doesNotOverwriteExistingErrorHandler() {
+        RxJavaPlugins.reset()
+        val existingHandler = Consumer<Throwable> { }
+        RxJavaPlugins.setErrorHandler(existingHandler)
+
+        ConstructorIo.setupRxJavaErrorHandler()
+
+        assertSame(existingHandler, RxJavaPlugins.getErrorHandler())
+    }
+
+    @Test
+    fun handlesUndeliverableIOException() {
+        // Should not throw - IOException is handled gracefully
+        RxJavaPlugins.onError(UndeliverableException(IOException("network timeout")))
+    }
+
+    @Test
+    fun handlesUndeliverableInterruptedException() {
+        // Should not throw - InterruptedException is handled gracefully
+        // and the interrupt flag should be restored on the current thread
+        Thread.interrupted() // clear any pre-existing interrupt flag
+        RxJavaPlugins.onError(UndeliverableException(InterruptedException("thread interrupted")))
+        assertTrue("Interrupt flag should be restored", Thread.interrupted())
+    }
+
+    @Test
+    fun handlesRawIOException() {
+        // Should not throw - IOException without UndeliverableException wrapper
+        RxJavaPlugins.onError(IOException("connection reset"))
+    }
+
+    @Test
+    fun handlesSocketTimeoutException() {
+        // Should not throw - SocketTimeoutException is a subclass of IOException
+        RxJavaPlugins.onError(UndeliverableException(SocketTimeoutException("connect timed out")))
+    }
+
+    @Test
+    fun forwardsUnexpectedExceptionToUncaughtExceptionHandler() {
+        val original = Thread.currentThread().uncaughtExceptionHandler
+        try {
+            var caughtThrowable: Throwable? = null
+            Thread.currentThread().uncaughtExceptionHandler =
+                Thread.UncaughtExceptionHandler { _, throwable -> caughtThrowable = throwable }
+
+            val error = IllegalStateException("unexpected error")
+            RxJavaPlugins.onError(UndeliverableException(error))
+
+            assertSame(error, caughtThrowable)
+        } finally {
+            Thread.currentThread().uncaughtExceptionHandler = original
+        }
+    }
+
+    @Test
+    fun logsUnexpectedExceptionWhenNoUncaughtHandler() {
+        val original = Thread.currentThread().uncaughtExceptionHandler
+        val defaultHandler = Thread.getDefaultUncaughtExceptionHandler()
+        try {
+            Thread.currentThread().uncaughtExceptionHandler = null
+            Thread.setDefaultUncaughtExceptionHandler(null)
+
+            // Should not throw — falls back to logging when both handlers are null
+            RxJavaPlugins.onError(UndeliverableException(IllegalStateException("no handler")))
+        } finally {
+            Thread.currentThread().uncaughtExceptionHandler = original
+            Thread.setDefaultUncaughtExceptionHandler(defaultHandler)
+        }
+    }
+
+    @Test
+    fun forwardsNullCauseExceptionToUncaughtExceptionHandler() {
+        val original = Thread.currentThread().uncaughtExceptionHandler
+        try {
+            var caughtThrowable: Throwable? = null
+            Thread.currentThread().uncaughtExceptionHandler =
+                Thread.UncaughtExceptionHandler { _, throwable -> caughtThrowable = throwable }
+
+            val error = UndeliverableException(null)
+            RxJavaPlugins.onError(error)
+
+            // When cause is null, the UndeliverableException itself is forwarded
+            assertSame(error, caughtThrowable)
+        } finally {
+            Thread.currentThread().uncaughtExceptionHandler = original
+        }
+    }
+}