Repository: arkivanov/Decompose (kotlin)

Issue:
On exception during navigation, other navigation events are not processed
I have discovered that if there is an exception during the navigation flow, e.g. serialization issue in `childFactory`, other attempts to navigate again are not processed.

Seems like [Relay](https://github.com/arkivanov/Decompose/blob/537dfbf9165b91a50e32fe468c06252c9f1d54a0/decompose/src/commonMain/kotlin/com/arkivanov/decompose/Relay.kt#L39) is stuck in `draining == true` - exception stops during the observer notification process, and `isDraining` is not set to `false`.

Sounds like a simple try/catch around observer notification solve the issue? If so, I can open a PR

```
try {
    observersCopy.forEach { observer ->
                observer(value)
    }
} catch (t: Throwable) {
   isDraining = false
   throw t
}
```

Pull request description:
Closes: #892

<!-- This is an auto-generated comment: release notes by coderabbit.ai -->

## Summary by CodeRabbit

* **Bug Fixes**
  * Improved error handling to ensure that if an error occurs during event processing, further events are blocked and the failure is signaled on subsequent attempts.

* **Tests**
  * Added comprehensive tests to verify correct behavior when errors occur during event dispatch, ensuring exceptions are propagated and no further processing occurs after a failure.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Patch:
diff --git a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/Relay.kt b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/Relay.kt
index 2e9e9e2f5..dfa7a7341 100644
--- a/decompose/src/commonMain/kotlin/com/arkivanov/decompose/Relay.kt
+++ b/decompose/src/commonMain/kotlin/com/arkivanov/decompose/Relay.kt
@@ -6,9 +6,14 @@ internal class Relay<T> {
     private val queue = ArrayDeque<T>()
     private var isDraining = false
     private var observers = emptySet<(T) -> Unit>()
+    private var error: Throwable? = null
 
     fun accept(value: T) {
         lock.synchronized {
+            error?.also {
+                throw IllegalStateException("Can't process the event due to a previous failure", it)
+            }
+
             queue.addLast(value)
 
             if (isDraining) {
@@ -36,8 +41,17 @@ internal class Relay<T> {
                 observersCopy = observers
             }
 
-            observersCopy.forEach { observer ->
-                observer(value)
+            try {
+                observersCopy.forEach { observer ->
+                    observer(value)
+                }
+            } catch (e: Throwable) {
+                lock.synchronized {
+                    queue.clear()
+                    isDraining = false
+                    error = e
+                    throw e
+                }
             }
         }
     }


Test patch:
diff --git a/decompose/src/commonTest/kotlin/com/arkivanov/decompose/RelayTest.kt b/decompose/src/commonTest/kotlin/com/arkivanov/decompose/RelayTest.kt
index b4f3cde13..9d1f50d81 100644
--- a/decompose/src/commonTest/kotlin/com/arkivanov/decompose/RelayTest.kt
+++ b/decompose/src/commonTest/kotlin/com/arkivanov/decompose/RelayTest.kt
@@ -2,6 +2,9 @@ package com.arkivanov.decompose
 
 import kotlin.test.Test
 import kotlin.test.assertContentEquals
+import kotlin.test.assertEquals
+import kotlin.test.assertFails
+import kotlin.test.assertFailsWith
 import kotlin.test.assertFalse
 
 @Suppress("TestFunctionName")
@@ -62,6 +65,103 @@ class RelayTest {
         assertContentEquals(listOf(1, 2), list)
     }
 
+    @Test
+    fun GIVEN_observer_throws_WHEN_accept_THEN_exception_thrown() {
+        val relay = Relay<Int>()
+        val error = Exception()
+        relay.subscribe { throw error }
+
+        val throwError = assertFails { relay.accept(1) }
+
+        assertEquals(error, throwError)
+    }
+
+    @Test
+    fun GIVEN_two_observers_and_first_throws_WHEN_accept_THEN_exception_thrown() {
+        val relay = Relay<Int>()
+        val error = Exception()
+        relay.subscribe { throw error }
+        relay.subscribe {}
+
+        val throwError = assertFails { relay.accept(1) }
+
+        assertEquals(error, throwError)
+    }
+
+    @Test
+    fun GIVEN_two_observers_and_first_throws_WHEN_accept_THEN_second_observer_not_called() {
+        val relay = Relay<Int>()
+        val error = Exception()
+        relay.subscribe { throw error }
+        var isSecondObserverCalled = false
+        relay.subscribe { isSecondObserverCalled = true }
+
+        runCatching { relay.accept(1) }
+
+        assertFalse(isSecondObserverCalled)
+    }
+
+    @Test
+    fun GIVEN_observer_sends_recursively_and_throws_WHEN_accept_THEN_exception_thrown() {
+        val relay = Relay<Int>()
+        val error = Exception()
+
+        relay.subscribe {
+            relay.accept(2)
+            throw error
+        }
+
+        val throwError = assertFails { relay.accept(1) }
+
+        assertEquals(error, throwError)
+    }
+
+    @Test
+    fun GIVEN_observer_sends_recursively_and_throws_WHEN_accept_THEN_observer_called_only_once() {
+        val relay = Relay<Int>()
+        val error = Exception()
+        val values = ArrayList<Int>()
+
+        relay.subscribe {
+            values += it
+            relay.accept(2)
+            throw error
+        }
+
+        runCatching { relay.accept(1) }
+
+        assertEquals(listOf(1), values)
+    }
+
+    @Test
+    fun GIVEN_observer_throws_WHEN_accept_second_time_THEN_exception_thrown_with_cause() {
+        val relay = Relay<Int>()
+        val error = Exception()
+        relay.subscribe { throw error }
+        runCatching { relay.accept(1) }
+
+        val thrownError = assertFailsWith<IllegalStateException> { relay.accept(2) }
+
+        assertEquals(error, thrownError.cause)
+    }
+
+    @Test
+    fun GIVEN_observer_throws_WHEN_accept_twice_THEN_observer_called_only_once() {
+        val relay = Relay<Int>()
+        val error = Exception()
+        var callCount = 0
+
+        relay.subscribe {
+            callCount++
+            throw error
+        }
+
+        runCatching { relay.accept(1) }
+        runCatching { relay.accept(2) }
+
+        assertEquals(1, callCount)
+    }
+
     private fun <T> Relay<T>.subscribe(): List<T> {
         val list = ArrayList<T>()
         subscribe { list += it }