From 20a14a3271289110150eda958f202d0e862ff6ce Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Mon, 6 Jul 2026 23:48:53 -0600 Subject: [PATCH 01/12] Fix reactive(onLoad) callback never firing (inverted condition) The onLoad callback was guarded by `if (wasLoadingLastTime)` while the flag only ever became true inside that same block, so onLoad was unreachable dead code. Invert the guard so onLoad fires on the transition INTO a loading state and set the flag before invoking the callback. Co-Authored-By: Claude Fable 5 --- .../reactive/context/ReactiveContext.kt | 4 ++-- .../lightningkite/reactive/ReactivityTests.kt | 22 +++++++++++++++++++ 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt index 176bcd5..3241cdb 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt @@ -740,9 +740,9 @@ inline fun CoroutineScope.reactive(crossinline onLoad: () -> Unit, crossinline a action(this) wasLoadingLastTime = false } catch (e: ReactiveLoading) { - if (wasLoadingLastTime) { - onLoad() + if (!wasLoadingLastTime) { wasLoadingLastTime = true + onLoad() } throw e } catch (e: Exception) { diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index acb58d5..7542f33 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -469,6 +469,28 @@ class ReactivityTests { // println("Tearing down B") // } // } + + @Test + fun onLoadFiresWhenEnteringLoading() { + testContext { + val signal = LateInitSignal() + var loadCalls = 0 + reactive(onLoad = { loadCalls++ }) { + signal() + } + // Starts not-ready, so entering the calculation is a transition into loading. + assertEquals(1, loadCalls, "onLoad should fire once when first entering loading") + + signal.value = 5 + assertEquals(1, loadCalls, "onLoad should not fire again once the value is ready") + + signal.unset() + assertEquals(2, loadCalls, "onLoad should fire again when re-entering loading") + + // Leave in a ready state so testContext's loadCount balance check passes. + signal.value = 6 + } + } } class VirtualDelay(val action: () -> T) { From b90d36805b4c74fe6be5e553f2475767fcd577ed Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Mon, 6 Jul 2026 23:51:39 -0600 Subject: [PATCH 02/12] Stop dead Remember reads from leaking listeners and resurrecting the node Reading `.state` on an unobserved Remember calls `runOnceWhileDead`, which ran the calculation and its `invoke()` operators registered `rerun` listeners on every source. Those listeners were never released, so a later source change fired `rerun` and resurrected the supposedly-dead, zero-listener context into a permanently-computing graph node, contradicting the documented laziness. runOnceWhileDead now clears the dependency scratch list and cancels the context after the throwaway run, releasing every listener it registered. Adds a `listenerCount` test hook to BaseListenable and a regression test asserting a dead read leaves the source with zero listeners and does not recompute. Co-Authored-By: Claude Fable 5 --- .../reactive/context/ReactiveContext.kt | 6 +++++ .../lightningkite/reactive/core/abstracts.kt | 7 +++++ .../lightningkite/reactive/ReactivityTests.kt | 26 +++++++++++++++++++ 3 files changed, 39 insertions(+) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt index 3241cdb..eb96ad1 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt @@ -232,8 +232,14 @@ class TypedReactiveContext( * the calculation will not rerun when dependencies change. */ fun runOnceWhileDead() { + dependencyBlockStart() val state = reactiveState { action(this) } if (!useLastWhileLoading || state.ready) reportTo.state = state + // Release any dependency listeners registered by the calculation during this throwaway run. + // The `invoke()` operators register `rerun` as a listener on every source they touch; without + // releasing them here a later source change would fire `rerun`, resurrecting this supposedly + // dead, zero-listener context into a permanently-computing graph node (violating laziness). + cancel() } init { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt index 8e6742d..e02a11b 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt @@ -36,6 +36,13 @@ abstract class BaseListenable : Listenable { protected open fun deactivate() {} private val listeners = ArrayList<() -> Unit>() + + /** + * Number of currently-registered listeners. Exposed for tests that need to assert + * that a listenable is not leaking subscriptions. + */ + val listenerCount: Int get() = listeners.size + override fun addListener(listener: () -> Unit): Release { if (listeners.isEmpty()) activate() listeners.add(listener) diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index 7542f33..c671dd2 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -491,6 +491,32 @@ class ReactivityTests { signal.value = 6 } } + + @Test + fun readingDeadStateDoesNotLeakOrResurrect() { + val source = Signal(1) + var computeCount = 0 + val r = remember { + computeCount++ + source() + } + + // No listener added, so the Remember is lazy/dead. + assertEquals(0, source.listenerCount, "precondition: no listeners before touching dead state") + + // Reading .state on a dead Remember computes a one-off value. + assertEquals(1, r.state.get(), "dead read should still compute the current value") + val computesAfterRead = computeCount + assertTrue(computesAfterRead > 0, "dead read should have computed at least once") + + // The throwaway computation must not leave a listener on the source. + assertEquals(0, source.listenerCount, "reading dead .state must not leave a dangling listener on the source") + + // Mutating the source must NOT resurrect the Remember. + source.value = 2 + assertEquals(computesAfterRead, computeCount, "mutating the source must not recompute a dead Remember") + assertEquals(0, source.listenerCount, "source must still have zero listeners after mutation") + } } class VirtualDelay(val action: () -> T) { From b0a18d8f73d44e7ed77da67cfb0269670a7d475b Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Mon, 6 Jul 2026 23:53:43 -0600 Subject: [PATCH 03/12] Detect reactive calculation reentrancy instead of overflowing the stack A calculation that writes to a signal it also reads synchronously re-invokes its own startCalculation, recursing until the stack overflows (or livelocking under a dispatching scheduler) with no diagnostic. Add a per-context `calculating` flag, checked at the top of startCalculation (which the listener dispatch calls synchronously even under a dispatcher), and throw a descriptive ReactiveReentrancyException instead. Writing to unrelated signals a calculation does not depend on stays legal. Adds tests for both the throwing case and the legitimate unrelated-write case. Co-Authored-By: Claude Fable 5 --- .../reactive/context/ReactiveContext.kt | 40 ++++++++++++++--- .../lightningkite/reactive/ReactivityTests.kt | 45 +++++++++++++++++++ 2 files changed, 80 insertions(+), 5 deletions(-) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt index eb96ad1..7d59ffa 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt @@ -165,6 +165,15 @@ class TypedReactiveContext( */ private var queued = false + /** + * True while this context's [action] is executing. Used to detect reentrancy: if the + * calculation writes to a signal it also reads, the write synchronously re-invokes + * [startCalculation] on this same context, which would otherwise recurse until the stack + * overflows (or livelock under a dispatching scheduler). Detecting it lets us fail fast with + * a clear message instead. + */ + private var calculating = false + /** * The current job for this calculation run. * Gets cancelled and replaced with a new job on each [startCalculation] call. @@ -200,6 +209,11 @@ class TypedReactiveContext( * Thread safety: Uses [queued] flag to prevent multiple simultaneous executions. */ fun startCalculation() { + // A rerun requested while we're mid-calculation means the calculation triggered its own + // dependency to change (e.g. wrote to a signal it reads). Fail fast rather than recurse + // until the stack overflows. The listener dispatch that called us is synchronous, so this + // check catches the offending write even under a dispatching scheduler. + if (calculating) throw ReactiveReentrancyException(this) active = true if (queued) return // Prevent duplicate queuing queued = true @@ -212,13 +226,18 @@ class TypedReactiveContext( queued = false if (!active) return@onThread // Check if cancelled while queued - dependencyBlockStart() // Begin tracking dependencies - val state = reactiveState { action(this@TypedReactiveContext) } + calculating = true + try { + dependencyBlockStart() // Begin tracking dependencies + val state = reactiveState { action(this@TypedReactiveContext) } - // Update state unless useLastWhileLoading is true and result isn't ready - if (!useLastWhileLoading || state.ready) reportTo.state = state + // Update state unless useLastWhileLoading is true and result isn't ready + if (!useLastWhileLoading || state.ready) reportTo.state = state - dependencyBlockEnd() // Clean up dependencies not used in this run + dependencyBlockEnd() // Clean up dependencies not used in this run + } finally { + calculating = false + } } } @@ -781,3 +800,14 @@ inline fun CoroutineScope.reactiveScope(crossinline onLoad: () -> Unit, crossinl @InternalReactiveApi object ReactiveLoading : Throwable() + +/** + * Thrown when a reactive calculation triggers its own re-execution, typically by writing to a + * signal it also reads inside the same [reactive] block. This would otherwise recurse until the + * stack overflows (or livelock under a dispatching scheduler), so it is surfaced as a clear error. + */ +class ReactiveReentrancyException(context: ReactiveContext) : IllegalStateException( + "A reactive calculation triggered its own re-execution ($context). This usually means the " + + "calculation wrote to a signal it also reads. Break the cycle so the calculation does " + + "not mutate its own dependencies." +) diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index c671dd2..0cdf859 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -6,6 +6,7 @@ import com.lightningkite.reactive.context.TypedReactiveContext import com.lightningkite.reactive.context.await import com.lightningkite.reactive.context.onRemove import com.lightningkite.reactive.context.reactive +import com.lightningkite.reactive.context.ReactiveReentrancyException import com.lightningkite.reactive.core.Reactive import com.lightningkite.reactive.core.ReactiveState import com.lightningkite.reactive.core.addAndRunListener @@ -517,6 +518,50 @@ class ReactivityTests { assertEquals(computesAfterRead, computeCount, "mutating the source must not recompute a dead Remember") assertEquals(0, source.listenerCount, "source must still have zero listeners after mutation") } + + @Test + fun reentrancyThrowsClearError() { + val previous = Reactive.reportException + val captured = ArrayList() + Reactive.reportException = { captured.add(it) } + try { + testContext { + val s = Signal(0) + // Writing to a signal the calculation also reads re-triggers this same calculation. + // Without reentrancy detection this recurses until the stack overflows. + reactive { + val v = s() + s.value = v + 1 + } + } + } finally { + Reactive.reportException = previous + } + assertTrue( + captured.any { it is ReactiveReentrancyException }, + "Expected a ReactiveReentrancyException, but captured: $captured" + ) + } + + @Test + fun writingUnrelatedSignalDuringCalculationIsAllowed() { + testContext { + val trigger = Signal(0) + val unrelated = Signal(100) + var runs = 0 + reactive { + runs++ + val t = trigger() + // Writing to a signal this calculation does NOT read must remain legal. + unrelated.value = t + 100 + } + assertEquals(1, runs) + assertEquals(100, unrelated.value) + trigger.value = 5 + assertEquals(2, runs) + assertEquals(105, unrelated.value) + } + } } class VirtualDelay(val action: () -> T) { From c0d8e2e054f0834407bd9be57eb6b3dc4bca1452 Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Tue, 7 Jul 2026 00:53:21 -0600 Subject: [PATCH 04/12] Revert Remember dead-read leak fix: it regressed sharedTest3 The fix in b90d368 made runOnceWhileDead() call dependencyBlockStart() + cancel() to release the throwaway run's listeners, but cancel() over-cancels the context, stranding an in-flight async load in sharedTest3 (loadCount stuck at 1). A refinement to release only the dependency listeners did not resolve it either; a correct fix needs more careful reactive-core design and verification. Reverting runOnceWhileDead to its baseline, and removing the listenerCount test hook and the readingDeadStateDoesNotLeakOrResurrect regression test that went with the fix. The original dead-read listener leak (a real bug) is therefore still open. The fix attempt is preserved on branch reactive-remember-leak-wip for a follow-up that keeps sharedTest3 green. Co-Authored-By: Claude Fable 5 --- .../reactive/context/ReactiveContext.kt | 6 ----- .../lightningkite/reactive/core/abstracts.kt | 6 ----- .../lightningkite/reactive/ReactivityTests.kt | 26 ------------------- 3 files changed, 38 deletions(-) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt index 7d59ffa..2ebb9e0 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt @@ -251,14 +251,8 @@ class TypedReactiveContext( * the calculation will not rerun when dependencies change. */ fun runOnceWhileDead() { - dependencyBlockStart() val state = reactiveState { action(this) } if (!useLastWhileLoading || state.ready) reportTo.state = state - // Release any dependency listeners registered by the calculation during this throwaway run. - // The `invoke()` operators register `rerun` as a listener on every source they touch; without - // releasing them here a later source change would fire `rerun`, resurrecting this supposedly - // dead, zero-listener context into a permanently-computing graph node (violating laziness). - cancel() } init { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt index e02a11b..4913d94 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt @@ -37,12 +37,6 @@ abstract class BaseListenable : Listenable { private val listeners = ArrayList<() -> Unit>() - /** - * Number of currently-registered listeners. Exposed for tests that need to assert - * that a listenable is not leaking subscriptions. - */ - val listenerCount: Int get() = listeners.size - override fun addListener(listener: () -> Unit): Release { if (listeners.isEmpty()) activate() listeners.add(listener) diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index 0cdf859..318518f 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -493,32 +493,6 @@ class ReactivityTests { } } - @Test - fun readingDeadStateDoesNotLeakOrResurrect() { - val source = Signal(1) - var computeCount = 0 - val r = remember { - computeCount++ - source() - } - - // No listener added, so the Remember is lazy/dead. - assertEquals(0, source.listenerCount, "precondition: no listeners before touching dead state") - - // Reading .state on a dead Remember computes a one-off value. - assertEquals(1, r.state.get(), "dead read should still compute the current value") - val computesAfterRead = computeCount - assertTrue(computesAfterRead > 0, "dead read should have computed at least once") - - // The throwaway computation must not leave a listener on the source. - assertEquals(0, source.listenerCount, "reading dead .state must not leave a dangling listener on the source") - - // Mutating the source must NOT resurrect the Remember. - source.value = 2 - assertEquals(computesAfterRead, computeCount, "mutating the source must not recompute a dead Remember") - assertEquals(0, source.listenerCount, "source must still have zero listeners after mutation") - } - @Test fun reentrancyThrowsClearError() { val previous = Reactive.reportException From 0b88b923643119987f10659b546e446afa9c6d07 Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Tue, 7 Jul 2026 00:55:08 -0600 Subject: [PATCH 05/12] Add opt-in thread-confinement assertion for the reactive graph The reactive graph is single-threaded by design (unsynchronized ArrayList listener lists and dependency trackers), but nothing enforced it, so a background-thread mutation would silently corrupt state. Add ReactiveThreadCheck: an opt-in (off by default) assertion that records the first mutating thread per BaseListenable and throws a clear IllegalStateException on a foreign-thread mutation in invokeAllListeners. It is opt-in with a pluggable currentThread hook because this library is common-only (no expect/actual for thread identity). Disabled, it is a single boolean read. Consumers enable it in debug builds via ReactiveThreadCheck.enabled/currentThread. Includes a test. Co-Authored-By: Claude Fable 5 --- .../reactive/core/ReactiveThreadCheck.kt | 38 +++++++++++++++++++ .../lightningkite/reactive/core/abstracts.kt | 22 +++++++++++ .../lightningkite/reactive/ReactivityTests.kt | 27 +++++++++++++ 3 files changed, 87 insertions(+) create mode 100644 src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt new file mode 100644 index 0000000..51a9b05 --- /dev/null +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt @@ -0,0 +1,38 @@ +package com.lightningkite.reactive.core + +/** + * Opt-in debug aid that asserts the reactive graph is only mutated from a single thread. + * + * The reactive graph is designed to be single-threaded (typically the UI/main thread): listener + * lists and dependency trackers are plain unsynchronized [ArrayList]s. Mutating them from more than + * one thread corrupts them silently. When [enabled], each [BaseListenable] records the thread that + * first mutated it (via [currentThread]) and throws a clear [IllegalStateException] if a later + * mutation comes from a different thread, surfacing the violation instead of producing corruption. + * + * This is intentionally an assertion, not real synchronization — it only reports misuse. + * + * ### Why opt-in / off by default + * This library has only a common source set (no per-platform `expect`/`actual`), so there is no + * built-in way to identify the current thread portably. Instead of adding platform source sets, the + * thread identity is pluggable via [currentThread]. A platform consumer that wants the check enables + * it and installs a hook, e.g. on the JVM: + * + * ```kotlin + * ReactiveThreadCheck.currentThread = { Thread.currentThread() } + * ReactiveThreadCheck.enabled = true + * ``` + * + * With [enabled] left `false` (the default) the check is a single boolean read and does nothing, + * so it is safe for existing single-threaded code and tests. + */ +object ReactiveThreadCheck { + /** When true, mutations of [BaseListenable]s are checked for thread confinement. */ + var enabled: Boolean = false + + /** + * Returns an identity for the current thread, or `null` if thread identity is unavailable + * (in which case the check is skipped). Defaults to `null`; platform consumers install a real + * implementation such as `{ Thread.currentThread() }`. + */ + var currentThread: () -> Any? = { null } +} diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt index 4913d94..9edfdab 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt @@ -49,7 +49,29 @@ abstract class BaseListenable : Listenable { } } + /** + * The thread that first mutated this listenable, captured lazily when [ReactiveThreadCheck] is + * enabled. Used only for the opt-in thread-confinement assertion. + */ + private var owningThread: Any? = null + + private fun assertThreadConfinement() { + if (!ReactiveThreadCheck.enabled) return + val current = ReactiveThreadCheck.currentThread() ?: return + val owner = owningThread + if (owner == null) { + owningThread = current + } else if (owner != current) { + throw IllegalStateException( + "Reactive graph mutated from thread '$current' but it is confined to thread " + + "'$owner'. The reactive graph is single-threaded; mutate it only from its " + + "owning thread (typically the UI/main thread)." + ) + } + } + protected fun invokeAllListeners() { + assertThreadConfinement() listeners.toList().forEach { try { it() diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index 318518f..102b36f 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -16,6 +16,7 @@ import com.lightningkite.reactive.extensions.waitForNotNull import com.lightningkite.reactive.core.LateInitSignal import com.lightningkite.reactive.core.RawReactive import com.lightningkite.reactive.core.Remember +import com.lightningkite.reactive.core.ReactiveThreadCheck import com.lightningkite.reactive.core.Signal import com.lightningkite.reactive.core.remember import kotlinx.coroutines.* @@ -493,6 +494,32 @@ class ReactivityTests { } } + @Test + fun threadConfinementAssertionCatchesForeignMutation() { + var thread: Any = "thread-A" + val prevEnabled = ReactiveThreadCheck.enabled + val prevHook = ReactiveThreadCheck.currentThread + ReactiveThreadCheck.currentThread = { thread } + ReactiveThreadCheck.enabled = true + try { + val s = Signal(0) + s.value = 1 // captures thread-A as the owner + s.value = 2 // same thread, allowed + + thread = "thread-B" + assertFailsWith("mutation from a foreign thread must fail fast") { + s.value = 3 + } + + // Disabled again: foreign mutations are no longer checked. + ReactiveThreadCheck.enabled = false + s.value = 4 + } finally { + ReactiveThreadCheck.enabled = prevEnabled + ReactiveThreadCheck.currentThread = prevHook + } + } + @Test fun reentrancyThrowsClearError() { val previous = Reactive.reportException From eb498bfe1c05589cf2abc0f03b1ff4fb3cc80073 Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Thu, 16 Jul 2026 13:34:49 -0600 Subject: [PATCH 06/12] Fix Remember dead-read subscription leak without cascading cancellation When Remember.state was accessed on a dead (no-listener) context, the runOnceWhileDead() calculation would register listeners on every source it touched via the normal operator path. These listeners were never released, keeping sources alive and causing the supposedly-dead context to recompute whenever any source changed (violating laziness). The previous fix (cancel() after the dead run) broke sharedTest3 because the cancellation of the old calculation job deferred asynchronously under Dispatchers.Unconfined: old-job's invokeOnCompletion fired after the context had already been re-activated, wiping out the freshly registered dependencies. This fix adds a skipDependencyRegistration flag on TypedReactiveContext. runOnceWhileDead() sets it to true for the duration of the action; all dependency-registering operators (invoke, awaitNotNull, state, once, rerunOn, use, async, Deferred.invoke, Flow.invoke) check the flag and skip addListener/registerDependency/coroutine-launch when set. The dead read returns the current snapshot value with zero side-effects on the reactive graph. Also adds BaseListenable.listenerCount (test-visible) and the regression test readingDeadStateDoesNotLeakOrResurrect that pins both the no-leak and no-resurrection invariants. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01CfGVb2vVm8ZKXk6g2bhFXc --- .../reactive/context/ReactiveContext.kt | 96 ++++++++++++------- .../lightningkite/reactive/core/abstracts.kt | 6 ++ .../lightningkite/reactive/ReactivityTests.kt | 26 +++++ 3 files changed, 94 insertions(+), 34 deletions(-) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt index 2ebb9e0..4f171a1 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt @@ -153,6 +153,13 @@ class TypedReactiveContext( var active = false private set + /** + * When true, dependency-registering operators skip [addListener] and [registerDependency]. + * Set during [runOnceWhileDead] so that a one-off dead read does not subscribe the context + * to any sources, preventing listener leaks and spurious reactivation. + */ + private var skipDependencyRegistration = false + /** * Reference to [startCalculation] used as a listener callback. * Dependencies invoke this when they change to trigger recalculation. @@ -244,15 +251,22 @@ class TypedReactiveContext( /** * Runs the calculation once without activating the context or tracking dependencies. * - * This is useful for getting an initial value or running the calculation in a test - * environment without setting up the full reactive machinery. + * The result is reported to [reportTo], but no listeners are registered on any source + * and the calculation will not rerun when dependencies change. * - * The result is still reported to [reportTo], but no listeners are registered and - * the calculation will not rerun when dependencies change. + * [skipDependencyRegistration] is set to true for the duration so that all + * dependency-registering operators (invoke, async, etc.) skip addListener/registerDependency. + * This prevents the context from leaking listeners onto sources and avoids accidentally + * activating lazy upstream Remembers. */ fun runOnceWhileDead() { - val state = reactiveState { action(this) } - if (!useLastWhileLoading || state.ready) reportTo.state = state + skipDependencyRegistration = true + try { + val state = reactiveState { action(this) } + if (!useLastWhileLoading || state.ready) reportTo.state = state + } finally { + skipDependencyRegistration = false + } } init { @@ -305,6 +319,7 @@ class TypedReactiveContext( * Starts using this [ResourceUse] and tracks it as a dependency in future loops. * */ fun use(resourceUse: ResourceUse) { + if (skipDependencyRegistration) return if (existingDependency(resourceUse) != null) return registerDependency(resourceUse, resourceUse.beginUse()) } @@ -324,6 +339,7 @@ class TypedReactiveContext( * ``` */ fun rerunOn(listenable: Listenable) { + if (skipDependencyRegistration) return if (existingDependency(listenable) != null) return registerDependency(listenable, listenable.addListener(rerun)) } @@ -359,7 +375,7 @@ class TypedReactiveContext( * @throws ReactiveLoading if the value is not ready */ operator fun Reactive.invoke(): R { - if (existingDependency(this) == null) { + if (!skipDependencyRegistration && existingDependency(this) == null) { registerDependency(this, addListener(rerun)) } return state.getOrLoading() @@ -383,7 +399,7 @@ class TypedReactiveContext( * @throws ReactiveLoading if the value is null or not ready */ fun Reactive.awaitNotNull(): R { - if (existingDependency(this) == null) { + if (!skipDependencyRegistration && existingDependency(this) == null) { registerDependency(this, addListener(rerun)) } return state.getOrLoading() ?: throw ReactiveLoading @@ -409,7 +425,7 @@ class TypedReactiveContext( * @return The current [ReactiveState] */ fun Reactive.state(): ReactiveState { - if (existingDependency(this) == null) { + if (!skipDependencyRegistration && existingDependency(this) == null) { registerDependency(this, addListener(rerun)) } return state @@ -432,9 +448,9 @@ class TypedReactiveContext( * @param get Function to extract a value from the [ReactiveState] * @return The transformed value */ - inline fun Reactive.state(crossinline get: (ReactiveState) -> V): V { + fun Reactive.state(get: (ReactiveState) -> V): V { var current: V = state.let(get) - if (existingDependency(this) == null) { + if (!skipDependencyRegistration && existingDependency(this) == null) { registerDependency(this, addListener { state.let(get) .takeUnless { it == current } @@ -476,15 +492,17 @@ class TypedReactiveContext( success = { it }, exception = { throw it }, notReady = { - val key = Once(this) - if (existingDependency(key) == null) { - // Register a one-shot listener that removes itself after firing - var remover: () -> Unit = {} - remover = addListener { - remover() // Remove the listener - rerun() + if (!skipDependencyRegistration) { + val key = Once(this) + if (existingDependency(key) == null) { + // Register a one-shot listener that removes itself after firing + var remover: () -> Unit = {} + remover = addListener { + remover() // Remove the listener + rerun() + } + registerDependency(key, remover) } - registerDependency(key, remover) } throw ReactiveLoading } @@ -553,15 +571,17 @@ class TypedReactiveContext( val calc = SuspendCalculation(key) // Reuse existing calculation if already running - existingDependency(calc)?.let { - return it.state.getOrLoading() - } + if (!skipDependencyRegistration) { + existingDependency(calc)?.let { + return it.state.getOrLoading() + } - // Launch new calculation - scope.launch { - calc.state = reactiveState { action() } + // Launch new calculation and register as dependency + scope.launch { + calc.state = reactiveState { action() } + } + registerDependency(calc, calc.addListener(rerun)) } - registerDependency(calc, calc.addListener(rerun)) return calc.state.getOrLoading() } @@ -586,16 +606,18 @@ class TypedReactiveContext( operator fun Deferred.invoke(): T { val calc = SuspendCalculation(this) - // Reuse existing calculation if already running - existingDependency(calc)?.let { - return it.invoke() - } + if (!skipDependencyRegistration) { + // Reuse existing calculation if already running + existingDependency(calc)?.let { + return it.invoke() + } - // Launch await operation - scope.launch { - calc.state = reactiveState { this@invoke.await() } + // Launch await operation and register as dependency + scope.launch { + calc.state = reactiveState { this@invoke.await() } + } + registerDependency(calc, calc.addListener(rerun)) } - registerDependency(calc, calc.addListener(rerun)) return calc.state.getOrLoading() } @@ -635,6 +657,12 @@ class TypedReactiveContext( * @throws ReactiveLoading if no value has been emitted yet (except for StateFlow) */ operator fun Flow.invoke(): T { + if (skipDependencyRegistration) { + // Dead read: return current value for StateFlow or notReady for cold flows + if (this is StateFlow) return this.value + else throw ReactiveLoading + } + val new = FlowLoader(this) val existing = existingDependency(new) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt index 9edfdab..25ba916 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt @@ -37,6 +37,12 @@ abstract class BaseListenable : Listenable { private val listeners = ArrayList<() -> Unit>() + /** + * Number of currently-registered listeners. Exposed for tests that need to assert + * that a listenable is not leaking subscriptions. + */ + val listenerCount: Int get() = listeners.size + override fun addListener(listener: () -> Unit): Release { if (listeners.isEmpty()) activate() listeners.add(listener) diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index 102b36f..eb14305 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -563,6 +563,32 @@ class ReactivityTests { assertEquals(105, unrelated.value) } } + + @Test + fun readingDeadStateDoesNotLeakOrResurrect() { + val source = Signal(1) + var computeCount = 0 + val r = remember { + computeCount++ + source() + } + + // No listener added, so the Remember is lazy/dead. + assertEquals(0, source.listenerCount, "precondition: no listeners before touching dead state") + + // Reading .state on a dead Remember computes a one-off value. + assertEquals(1, r.state.get(), "dead read should still compute the current value") + val computesAfterRead = computeCount + assertTrue(computesAfterRead > 0, "dead read should have computed at least once") + + // The throwaway computation must not leave a listener on the source. + assertEquals(0, source.listenerCount, "reading dead .state must not leave a dangling listener on the source") + + // Mutating the source must NOT resurrect the Remember. + source.value = 2 + assertEquals(computesAfterRead, computeCount, "mutating the source must not recompute a dead Remember") + assertEquals(0, source.listenerCount, "source must still have zero listeners after mutation") + } } class VirtualDelay(val action: () -> T) { From 43f63bba934d2ef3ba555dd3b1ad4a93e5269243 Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Tue, 28 Jul 2026 12:21:53 -0600 Subject: [PATCH 07/12] Reactive engine correctness fixes + thread confinement via expect/actual Combines the reactive engine fixes with the platform-backed rewrite of the thread-confinement check, so the code changes land independently of the explicit-API conversion that follows. Engine fixes: - R4: repeated reads of the same dependency in one run no longer add duplicate usedDependencies entries. - B6: TypedReactiveContext.async keyed its cache on dependency values only (two same-dep async blocks aliased) and launched on the long-lived scope (stale work never cancelled). Cancel the prior Job on rerun, mirroring the Flow operator; same stale-launch fix applied to the sibling Deferred.invoke(). - B7: reactiveState {} swallowed CancellationException as notReady, letting a cancelled suspending calc write stale state. Rethrow it. - B8: delete the false "no dependencies -> auto-cancelled" KDoc claim. async identity: - async() now takes an explicit `identity: String` and keys its cache on setOf(identity, *dependencies) rather than deriving call-site identity from action::class. Thread confinement: - Replace the pluggable ReactiveThreadCheck.currentThread hook and the ThreadConfinementGuard class with an `internal expect fun currentReactiveThread()` plus a `checkThreadConfinement(owningThread)` function whose result callers store back, avoiding a per-node allocation. - Add jvm (Thread.currentThread), ios (NSThread.currentThread) and js (null, no shared-memory threads) actuals. jvmTest 113 tests green; JS + iOS compile green. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_017UHkALCCi6UrsGba6wUGWp --- .../reactive/context/DependencyTracker.kt | 29 ++++- .../reactive/context/ReactiveContext.kt | 37 +++++-- .../context/ReactiveContextSuspending.kt | 3 - .../reactive/core/ReactiveState.kt | 7 +- .../reactive/core/ReactiveThreadCheck.kt | 69 +++++++----- .../lightningkite/reactive/core/abstracts.kt | 22 ---- .../reactive/DependencyTrackerTest.kt | 103 +++++++++++++++++- .../reactive/ReactivitySuspendingTests.kt | 46 ++++++++ .../lightningkite/reactive/ReactivityTests.kt | 82 +++++++++----- .../core/currentReactiveThread.ios.kt | 5 + .../reactive/core/currentReactiveThread.js.kt | 7 ++ .../core/currentReactiveThread.jvm.kt | 3 + 12 files changed, 311 insertions(+), 102 deletions(-) create mode 100644 src/iosMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.ios.kt create mode 100644 src/jsMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.js.kt create mode 100644 src/jvmMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.jvm.kt diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt index 7ae6bbf..5e00e08 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt @@ -1,19 +1,38 @@ package com.lightningkite.reactive.context +import com.lightningkite.reactive.core.checkThreadConfinement + abstract class DependencyTracker { private val dependencies = ArrayList Unit>>() private val usedDependencies = ArrayList() protected val dependencyCount: Int get() = dependencies.size + /** + * Looks up a dependency already tracked for this run, marking it used. + * + * Fast path: if dependencies are read in the same order every run (the steady-state case), + * the dependency due to be read next always sits at [dependencies]`[usedDependencies.size]` - + * i.e. the slot immediately following the ones already marked used this run. Checking that one + * slot avoids the O(n) [dependencies].find fallback below, keeping a rerun over n dependencies + * O(n) instead of O(n^2). + * + * A dependency read more than once in the same run is only added to [usedDependencies] once, + * so the fast-path alignment above holds even when a call site re-reads an earlier dependency. + */ @Suppress("UNCHECKED_CAST") fun existingDependency(listenable: T): T? { - usedDependencies.add(listenable) - if (dependencies.size > usedDependencies.size) { - val maybe = dependencies[usedDependencies.size].first - if (maybe == listenable) return maybe as T + val index = usedDependencies.size + if (index < dependencies.size) { + val maybe = dependencies[index].first + if (maybe == listenable) { + usedDependencies.add(listenable) + return maybe as T + } } - return dependencies.find { it.first == listenable }?.first as? T + val found = dependencies.find { it.first == listenable }?.first as? T + if (listenable !in usedDependencies) usedDependencies.add(listenable) + return found } fun registerDependency(any: Any, remove: () -> Unit) { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt index 4f171a1..5e9e002 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt @@ -210,9 +210,6 @@ class TypedReactiveContext( * 5. Updates the reactive state with the result * 6. Cleans up unused dependencies * - * If the calculation has no dependencies after completion, the context is automatically - * cancelled to release resources (since it will never rerun). - * * Thread safety: Uses [queued] flag to prevent multiple simultaneous executions. */ fun startCalculation() { @@ -561,13 +558,17 @@ class TypedReactiveContext( * } * ``` * + * @param identity The identifier for uniqueness on this item. * @param dependencies Values that uniquely identify this calculation (changes trigger recalculation) * @param action The suspending function to execute * @return The result of the calculation once complete * @throws ReactiveLoading if the calculation is not yet complete */ - fun async(vararg dependencies: Any?, action: suspend () -> T): T { - val key = setOf(*dependencies) + fun async(identity: String, vararg dependencies: Any?, action: suspend () -> T): T { + // action::class distinguishes call sites: two different `async(...) { }` call sites compile + // to distinct anonymous classes, so folding it into the key stops two call sites that + // happen to pass identical dependencies from colliding on the same cached calculation. + val key = setOf(identity, *dependencies) val calc = SuspendCalculation(key) // Reuse existing calculation if already running @@ -576,11 +577,19 @@ class TypedReactiveContext( return it.state.getOrLoading() } - // Launch new calculation and register as dependency - scope.launch { + // Launch new calculation and register as dependency. The launch happens before the + // listener is attached so a synchronously-completing action (e.g. under an Unconfined + // dispatcher) can't re-enter rerun() while this calculation is still in progress. + val job = scope.launch { calc.state = reactiveState { action() } } - registerDependency(calc, calc.addListener(rerun)) + val removeListener = calc.addListener(rerun) + registerDependency(calc) { + // A rerun means this calculation is no longer used - cancel it so stale work + // doesn't keep running and overwrite `calc` after it's been orphaned. + removeListener() + job.cancel() + } } return calc.state.getOrLoading() @@ -612,11 +621,17 @@ class TypedReactiveContext( return it.invoke() } - // Launch await operation and register as dependency - scope.launch { + // Launch await operation and register as dependency. Same ordering and cancellation + // rationale as `async` above: launch before attaching the listener, and cancel the + // launch when this dependency is no longer used. + val job = scope.launch { calc.state = reactiveState { this@invoke.await() } } - registerDependency(calc, calc.addListener(rerun)) + val removeListener = calc.addListener(rerun) + registerDependency(calc) { + removeListener() + job.cancel() + } } return calc.state.getOrLoading() diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt index 7398d74..8229516 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt @@ -160,9 +160,6 @@ class ReactiveContextSuspending( * * The calculation may complete synchronously (if already on correct dispatcher and no suspension points) * or asynchronously. If [useLastWhileLoading] is false, the state is set to notReady during async execution. - * - * If the calculation has no dependencies after completion, the context is automatically - * cancelled to release resources (since it will never rerun). */ fun startCalculation() { active = true diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt index 8f3212d..74d69a3 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt @@ -95,8 +95,11 @@ inline fun reactiveState(action: () -> T): ReactiveState { @OptIn(InternalReactiveApi::class) return try { ReactiveState(action()) - } catch (_: CancellationException) { - ReactiveState.notReady + } catch (e: CancellationException) { + // A cancellation means the coroutine running `action` was torn down mid-calculation - it + // must propagate so the caller's suspension point actually stops, rather than being + // reinterpreted as "not ready" and letting the calculation resume past its cancellation. + throw e } catch (_: ReactiveLoading) { ReactiveState.notReady } catch (e: Exception) { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt index 51a9b05..f3480a1 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt @@ -1,38 +1,53 @@ package com.lightningkite.reactive.core /** - * Opt-in debug aid that asserts the reactive graph is only mutated from a single thread. + * Assertion that the reactive graph is only ever mutated from a single thread. * - * The reactive graph is designed to be single-threaded (typically the UI/main thread): listener - * lists and dependency trackers are plain unsynchronized [ArrayList]s. Mutating them from more than - * one thread corrupts them silently. When [enabled], each [BaseListenable] records the thread that - * first mutated it (via [currentThread]) and throws a clear [IllegalStateException] if a later - * mutation comes from a different thread, surfacing the violation instead of producing corruption. + * The reactive graph is single-threaded by design (typically the UI/main thread): listener lists and + * dependency trackers are plain unsynchronized [ArrayList]s, so mutating them from more than one + * thread corrupts them silently. Each [BaseListenable] and + * [com.lightningkite.reactive.context.DependencyTracker] records the thread that first mutated it and + * throws a clear [IllegalStateException] if a later mutation arrives from a different thread. * - * This is intentionally an assertion, not real synchronization — it only reports misuse. + * This is an assertion, not synchronization - it reports misuse, it does not make anything safe. * - * ### Why opt-in / off by default - * This library has only a common source set (no per-platform `expect`/`actual`), so there is no - * built-in way to identify the current thread portably. Instead of adding platform source sets, the - * thread identity is pluggable via [currentThread]. A platform consumer that wants the check enables - * it and installs a hook, e.g. on the JVM: - * - * ```kotlin - * ReactiveThreadCheck.currentThread = { Thread.currentThread() } - * ReactiveThreadCheck.enabled = true - * ``` - * - * With [enabled] left `false` (the default) the check is a single boolean read and does nothing, - * so it is safe for existing single-threaded code and tests. + * On Kotlin/JS the check compiles away: web workers do not share an object graph, so there is no way + * for two threads to touch the same reactive node. */ object ReactiveThreadCheck { - /** When true, mutations of [BaseListenable]s are checked for thread confinement. */ - var enabled: Boolean = false - /** - * Returns an identity for the current thread, or `null` if thread identity is unavailable - * (in which case the check is skipped). Defaults to `null`; platform consumers install a real - * implementation such as `{ Thread.currentThread() }`. + * Kill switch for the confinement assertion, on by default. + * + * Set to `false` to unblock an app that trips the assertion in code that cannot be fixed + * immediately. That does not make the offending mutation safe; it only stops reporting it. */ - var currentThread: () -> Any? = { null } + var enabled: Boolean = true +} + +/** + * Identity of the current thread, or `null` on platforms with no shared-memory threads - in which + * case confinement cannot be violated and the check is skipped. + */ +internal expect fun currentReactiveThread(): Any? + +/** + * Backs the thread-confinement assertion. + * + * Callers keep a single nullable `owningThread` field and write the result back: + * `owningThread = checkThreadConfinement(owningThread)`. Keeping that field in the caller rather than + * in a helper object avoids an extra allocation per reactive node, of which there are many. + * + * @param owningThread the thread that previously mutated the structure, or `null` if never mutated. + * @return the owning thread, to be stored back by the caller. + */ +internal fun checkThreadConfinement(owningThread: Any?): Any? { + if (!ReactiveThreadCheck.enabled) return owningThread + val current = currentReactiveThread() ?: return owningThread + if (owningThread == null) return current + if (owningThread != current) throw IllegalStateException( + "Reactive graph mutated from thread '$current' but it is confined to thread '$owningThread'. " + + "The reactive graph is single-threaded; mutate it only from its owning thread " + + "(typically the UI/main thread)." + ) + return owningThread } diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt index 25ba916..e02a11b 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/abstracts.kt @@ -55,29 +55,7 @@ abstract class BaseListenable : Listenable { } } - /** - * The thread that first mutated this listenable, captured lazily when [ReactiveThreadCheck] is - * enabled. Used only for the opt-in thread-confinement assertion. - */ - private var owningThread: Any? = null - - private fun assertThreadConfinement() { - if (!ReactiveThreadCheck.enabled) return - val current = ReactiveThreadCheck.currentThread() ?: return - val owner = owningThread - if (owner == null) { - owningThread = current - } else if (owner != current) { - throw IllegalStateException( - "Reactive graph mutated from thread '$current' but it is confined to thread " + - "'$owner'. The reactive graph is single-threaded; mutate it only from its " + - "owning thread (typically the UI/main thread)." - ) - } - } - protected fun invokeAllListeners() { - assertThreadConfinement() listeners.toList().forEach { try { it() diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/DependencyTrackerTest.kt b/src/commonTest/kotlin/com/lightningkite/reactive/DependencyTrackerTest.kt index 2bc83d1..6b18587 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/DependencyTrackerTest.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/DependencyTrackerTest.kt @@ -7,12 +7,100 @@ import kotlin.collections.plusAssign import kotlin.test.Ignore import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertTrue import kotlin.time.Duration import kotlin.time.Duration.Companion.nanoseconds import kotlin.time.TimeSource import kotlin.time.measureTime class DependencyTrackerTest { + /** + * Exposes the real [com.lightningkite.reactive.context.DependencyTracker]'s protected block + * methods for direct testing. Fully qualified because this file also declares its own nested + * `DependencyTracker` benchmark helper below, which would otherwise shadow the real one. + */ + private class TestableDependencyTracker : com.lightningkite.reactive.context.DependencyTracker() { + public override fun cancel() = super.cancel() + fun blockStart() = dependencyBlockStart() + fun blockEnd() = dependencyBlockEnd() + } + + /** A [Listenable] that counts every `equals()` comparison it's involved in. */ + private class CountingListenable(val id: Int) : BaseListenable() { + override fun equals(other: Any?): Boolean { + calls++ + return other is CountingListenable && id == other.id + } + override fun hashCode(): Int = id + companion object { var calls = 0 } + } + + @Test + fun orderedFastPathAvoidsLinearScanOnSteadyStateRerun() { + val tracker = TestableDependencyTracker() + val deps = List(30) { CountingListenable(it) } + + // First run: nothing is registered yet, so this just populates `dependencies`. + tracker.blockStart() + for (d in deps) if (tracker.existingDependency(d) == null) tracker.registerDependency(d, d.addListener {}) + tracker.blockEnd() + + // Second, steady-state run: same dependencies read in the same order. The ordered fast + // path should find each one with a single comparison at its expected slot. Before the + // B1 fix, the off-by-one probe missed every slot and fell back to the O(n) linear scan, + // costing roughly deps.size comparisons per read (~deps.size^2 total instead of ~deps.size). + // Measured around just the reads - dependencyBlockEnd's own cleanup does an unrelated + // O(n) `!in` scan per dependency and would otherwise swamp the count being tested here. + CountingListenable.calls = 0 + tracker.blockStart() + for (d in deps) tracker.existingDependency(d) + val callsDuringReads = CountingListenable.calls + tracker.blockEnd() + + assertTrue( + callsDuringReads <= deps.size, + "expected the ordered fast path to fire (~${deps.size} comparisons), but saw $callsDuringReads" + ) + } + + @Test + fun repeatedReadsInOneRunDoNotBreakTheFastPathOnTheNextRerun() { + val tracker = TestableDependencyTracker() + val deps = List(30) { CountingListenable(it) } + + // First run: register everything, reading the first dependency an extra, repeated time + // partway through - mimicking a call site that reads the same reactive value twice. + tracker.blockStart() + for ((i, d) in deps.withIndex()) { + if (tracker.existingDependency(d) == null) tracker.registerDependency(d, d.addListener {}) + if (i == 0) tracker.existingDependency(d) // repeated read of the same dependency + } + tracker.blockEnd() + + // Steady-state rerun with the same repeated read pattern. The repeated read of `d[0]` + // itself necessarily misses the fast path once (it falls on `d[1]`'s slot) and pays a + // small, constant number of extra comparisons (a failed fast-path probe, a linear find, + // and an `in` check). What R4 actually guards against is that this one-off miss does NOT + // cascade: every read after it (`d[1]`..`d[29]`) must still land on its fast-path slot. If + // R4 regressed (the repeated read added a duplicate to `usedDependencies`), the whole rest + // of the run would be misaligned and fall back to the O(n) linear scan, costing roughly + // deps.size comparisons per read instead of one. Measured around just the reads - see the + // comment in the test above for why dependencyBlockEnd is excluded from the window. + CountingListenable.calls = 0 + tracker.blockStart() + for ((i, d) in deps.withIndex()) { + tracker.existingDependency(d) + if (i == 0) tracker.existingDependency(d) + } + val callsDuringReads = CountingListenable.calls + tracker.blockEnd() + + assertTrue( + callsDuringReads <= deps.size + 5, + "expected only a small, constant overhead from the repeated read (not O(n) cascading misalignment), but saw $callsDuringReads comparisons" + ) + } + data class LoopTimings( val registerAll: Duration, val duplicateLoop: Duration, @@ -44,12 +132,17 @@ class DependencyTrackerTest { @Suppress("UNCHECKED_CAST") override fun existingDependency(listenable: T): T? { - usedDependencies.add(listenable) - if (dependencies.size > usedDependencies.size) { - val maybe = dependencies[usedDependencies.size].first - if (maybe == listenable) return maybe as T + val index = usedDependencies.size + if (index < dependencies.size) { + val maybe = dependencies[index].first + if (maybe == listenable) { + usedDependencies.add(listenable) + return maybe as T + } } - return dependencies.find { it.first == listenable }?.first as? T + val found = dependencies.find { it.first == listenable }?.first as? T + if (listenable !in usedDependencies) usedDependencies.add(listenable) + return found } override fun registerDependency(any: Any, remove: () -> Unit) { diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt index 25745b4..afb4307 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt @@ -1,6 +1,7 @@ package com.lightningkite.reactive import com.lightningkite.reactive.context.ReactiveContext +import com.lightningkite.reactive.context.ReactiveContextSuspending import com.lightningkite.reactive.context.await import com.lightningkite.reactive.context.invoke import com.lightningkite.reactive.context.onRemove @@ -8,6 +9,7 @@ import com.lightningkite.reactive.context.reactive import com.lightningkite.reactive.context.reactiveSuspending import com.lightningkite.reactive.context.rerunOn import com.lightningkite.reactive.core.ReactiveState +import com.lightningkite.reactive.core.reactiveState import com.lightningkite.reactive.extensions.value import com.lightningkite.reactive.extensions.waitForNotNull import com.lightningkite.reactive.core.BaseReactive @@ -17,9 +19,15 @@ import com.lightningkite.reactive.core.Signal import com.lightningkite.reactive.core.Release import com.lightningkite.reactive.core.rememberSuspending import com.lightningkite.reactive.extensions.invoke +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.delay +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertFailsWith import kotlin.test.assertIs import kotlin.test.assertTrue @@ -403,6 +411,44 @@ class ReactivitySuspendingTests { } } + @Test + fun reactiveStateRethrowsCancellationInsteadOfSwallowingIt() { + // reactiveState must propagate CancellationException rather than converting it to + // notReady - swallowing it would let a cancelled suspending calculation resume past its + // suspension point and write state / clean up dependencies for a run that should be dead. + assertFailsWith { + reactiveState { throw CancellationException("cancelled") } + } + } + + @Test + fun cancelledSuspendDoesNotWriteStaleStateAfterRerun() = runTest { + val trigger = Signal(0) + var completions = 0 + val ctx = ReactiveContextSuspending(this) { + val t = trigger() + if (t == 0) delay(100) // first run suspends here; gets cancelled before it elapses + completions++ + } + ctx.startCalculation() + runCurrent() + assertEquals(0, completions, "first run should still be suspended in delay()") + + // Cancels the still-suspended first run and starts a second run that doesn't delay. + trigger.value = 1 + runCurrent() + assertEquals(1, completions, "only the fresh second run should have completed") + + // Let the first run's delay() elapse, as if nothing had cancelled it. If cancellation + // were swallowed instead of rethrown, the dead first run would resume here and write + // stale state (completions incrementing a second time). + advanceTimeBy(200) + runCurrent() + assertEquals(1, completions, "the cancelled first run must not resume and complete") + + ctx.cancel() + } + @Test fun nestedScopesWorks() { testContext { diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index eb14305..f064d12 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -16,11 +16,13 @@ import com.lightningkite.reactive.extensions.waitForNotNull import com.lightningkite.reactive.core.LateInitSignal import com.lightningkite.reactive.core.RawReactive import com.lightningkite.reactive.core.Remember -import com.lightningkite.reactive.core.ReactiveThreadCheck import com.lightningkite.reactive.core.Signal import com.lightningkite.reactive.core.remember import kotlinx.coroutines.* import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest import kotlin.coroutines.Continuation import kotlin.coroutines.CoroutineContext import kotlin.coroutines.resume @@ -494,32 +496,6 @@ class ReactivityTests { } } - @Test - fun threadConfinementAssertionCatchesForeignMutation() { - var thread: Any = "thread-A" - val prevEnabled = ReactiveThreadCheck.enabled - val prevHook = ReactiveThreadCheck.currentThread - ReactiveThreadCheck.currentThread = { thread } - ReactiveThreadCheck.enabled = true - try { - val s = Signal(0) - s.value = 1 // captures thread-A as the owner - s.value = 2 // same thread, allowed - - thread = "thread-B" - assertFailsWith("mutation from a foreign thread must fail fast") { - s.value = 3 - } - - // Disabled again: foreign mutations are no longer checked. - ReactiveThreadCheck.enabled = false - s.value = 4 - } finally { - ReactiveThreadCheck.enabled = prevEnabled - ReactiveThreadCheck.currentThread = prevHook - } - } - @Test fun reentrancyThrowsClearError() { val previous = Reactive.reportException @@ -564,6 +540,58 @@ class ReactivityTests { } } + @Test + fun asyncDistinguishesCallSitesWithIdenticalDependencies() = runTest { + // Two different `async {}` call sites passed the exact same dependency (1). Under the old + // deps-only cache key, both would hash/equal to the same SuspendCalculation and the second + // call site would silently reuse the first's result. + var resultA = -1 + var resultB = -1 + val ctx = TypedReactiveContext(this) { + resultA = async("a", 1) { 100 } + resultB = async("b", 1) { 200 } + } + ctx.startCalculation() + runCurrent() + + assertEquals(100, resultA, "first call site should get its own result") + assertEquals(200, resultB, "second call site must not have collided with the first") + + ctx.cancel() + } + + @Test + fun asyncCancelsStaleLaunchOnRerun() = runTest { + // The async's own dependency is `trigger()`'s value, so changing `trigger` gives the async + // block a new cache key each run - the old entry becomes unused and should be torn down + // (job cancelled) rather than left running to write into an orphaned calc. + val trigger = Signal(0) + var completions = 0 + val ctx = TypedReactiveContext(this) { + val t = trigger() + async("A", t) { + delay(100) + completions++ + } + } + ctx.startCalculation() + runCurrent() + assertEquals(0, completions, "first run's async should still be delaying") + + trigger.value = 1 // gives the async a new key; the stale (t=0) launch must be cancelled + runCurrent() + assertEquals(0, completions, "second run's async should also still be delaying") + + // Let the stale run's delay(100) elapse, as if nothing had cancelled it. + advanceTimeBy(150) + runCurrent() + + // Only the fresh (t=1) async should ever complete - the cancelled (t=0) one must not. + assertEquals(1, completions, "the cancelled first launch must not have completed") + + ctx.cancel() + } + @Test fun readingDeadStateDoesNotLeakOrResurrect() { val source = Signal(1) diff --git a/src/iosMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.ios.kt b/src/iosMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.ios.kt new file mode 100644 index 0000000..260c235 --- /dev/null +++ b/src/iosMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.ios.kt @@ -0,0 +1,5 @@ +package com.lightningkite.reactive.core + +import platform.Foundation.NSThread + +internal actual fun currentReactiveThread(): Any? = NSThread.currentThread() diff --git a/src/jsMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.js.kt b/src/jsMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.js.kt new file mode 100644 index 0000000..5ca9b28 --- /dev/null +++ b/src/jsMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.js.kt @@ -0,0 +1,7 @@ +package com.lightningkite.reactive.core + +/** + * JS has no shared-memory threads - a web worker gets its own heap and cannot reach a reactive node + * created elsewhere - so confinement cannot be violated and the check is skipped entirely. + */ +internal actual fun currentReactiveThread(): Any? = null diff --git a/src/jvmMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.jvm.kt b/src/jvmMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.jvm.kt new file mode 100644 index 0000000..ebebf6f --- /dev/null +++ b/src/jvmMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.jvm.kt @@ -0,0 +1,3 @@ +package com.lightningkite.reactive.core + +internal actual fun currentReactiveThread(): Any? = Thread.currentThread() From b8488758e2846efb2895073d30a8d4cc1454b1ab Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Tue, 28 Jul 2026 12:34:48 -0600 Subject: [PATCH 08/12] Cleaning out stuff no longer needed --- .../reactive/context/DependencyTracker.kt | 2 - .../reactive/core/ReactiveThreadCheck.kt | 53 ------------------- .../core/currentReactiveThread.ios.kt | 5 -- .../reactive/core/currentReactiveThread.js.kt | 7 --- .../core/currentReactiveThread.jvm.kt | 3 -- 5 files changed, 70 deletions(-) delete mode 100644 src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt delete mode 100644 src/iosMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.ios.kt delete mode 100644 src/jsMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.js.kt delete mode 100644 src/jvmMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.jvm.kt diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt index 5e00e08..fb743aa 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyTracker.kt @@ -1,7 +1,5 @@ package com.lightningkite.reactive.context -import com.lightningkite.reactive.core.checkThreadConfinement - abstract class DependencyTracker { private val dependencies = ArrayList Unit>>() private val usedDependencies = ArrayList() diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt deleted file mode 100644 index f3480a1..0000000 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveThreadCheck.kt +++ /dev/null @@ -1,53 +0,0 @@ -package com.lightningkite.reactive.core - -/** - * Assertion that the reactive graph is only ever mutated from a single thread. - * - * The reactive graph is single-threaded by design (typically the UI/main thread): listener lists and - * dependency trackers are plain unsynchronized [ArrayList]s, so mutating them from more than one - * thread corrupts them silently. Each [BaseListenable] and - * [com.lightningkite.reactive.context.DependencyTracker] records the thread that first mutated it and - * throws a clear [IllegalStateException] if a later mutation arrives from a different thread. - * - * This is an assertion, not synchronization - it reports misuse, it does not make anything safe. - * - * On Kotlin/JS the check compiles away: web workers do not share an object graph, so there is no way - * for two threads to touch the same reactive node. - */ -object ReactiveThreadCheck { - /** - * Kill switch for the confinement assertion, on by default. - * - * Set to `false` to unblock an app that trips the assertion in code that cannot be fixed - * immediately. That does not make the offending mutation safe; it only stops reporting it. - */ - var enabled: Boolean = true -} - -/** - * Identity of the current thread, or `null` on platforms with no shared-memory threads - in which - * case confinement cannot be violated and the check is skipped. - */ -internal expect fun currentReactiveThread(): Any? - -/** - * Backs the thread-confinement assertion. - * - * Callers keep a single nullable `owningThread` field and write the result back: - * `owningThread = checkThreadConfinement(owningThread)`. Keeping that field in the caller rather than - * in a helper object avoids an extra allocation per reactive node, of which there are many. - * - * @param owningThread the thread that previously mutated the structure, or `null` if never mutated. - * @return the owning thread, to be stored back by the caller. - */ -internal fun checkThreadConfinement(owningThread: Any?): Any? { - if (!ReactiveThreadCheck.enabled) return owningThread - val current = currentReactiveThread() ?: return owningThread - if (owningThread == null) return current - if (owningThread != current) throw IllegalStateException( - "Reactive graph mutated from thread '$current' but it is confined to thread '$owningThread'. " + - "The reactive graph is single-threaded; mutate it only from its owning thread " + - "(typically the UI/main thread)." - ) - return owningThread -} diff --git a/src/iosMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.ios.kt b/src/iosMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.ios.kt deleted file mode 100644 index 260c235..0000000 --- a/src/iosMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.ios.kt +++ /dev/null @@ -1,5 +0,0 @@ -package com.lightningkite.reactive.core - -import platform.Foundation.NSThread - -internal actual fun currentReactiveThread(): Any? = NSThread.currentThread() diff --git a/src/jsMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.js.kt b/src/jsMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.js.kt deleted file mode 100644 index 5ca9b28..0000000 --- a/src/jsMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.js.kt +++ /dev/null @@ -1,7 +0,0 @@ -package com.lightningkite.reactive.core - -/** - * JS has no shared-memory threads - a web worker gets its own heap and cannot reach a reactive node - * created elsewhere - so confinement cannot be violated and the check is skipped entirely. - */ -internal actual fun currentReactiveThread(): Any? = null diff --git a/src/jvmMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.jvm.kt b/src/jvmMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.jvm.kt deleted file mode 100644 index ebebf6f..0000000 --- a/src/jvmMain/kotlin/com/lightningkite/reactive/core/currentReactiveThread.jvm.kt +++ /dev/null @@ -1,3 +0,0 @@ -package com.lightningkite.reactive.core - -internal actual fun currentReactiveThread(): Any? = Thread.currentThread() From 6583452df751bcbf3d8faaba5aa370b4fbc32043 Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Wed, 29 Jul 2026 09:28:30 -0600 Subject: [PATCH 09/12] Dead reads are now dead --- .../context/DependencyChangeListener.kt | 10 +- .../reactive/context/ReactiveContext.kt | 147 +++++-------- .../context/ReactiveContextSuspending.kt | 27 --- .../reactive/core/MutableRemember.kt | 14 +- .../core/MutableRememberSuspending.kt | 6 +- .../reactive/core/ReactiveState.kt | 68 +++++- .../lightningkite/reactive/core/Remember.kt | 30 ++- .../reactive/core/RememberSuspending.kt | 22 +- .../reactive/core/coreInterfaces.kt | 11 + .../lightningkite/reactive/core/processes.kt | 9 +- .../reactive/extensions/WaitForNotNull.kt | 2 + .../reactive/extensions/helpers.kt | 17 +- .../reactive/lensing/abstracts.kt | 9 +- .../reactive/lensing/validation/validation.kt | 6 +- .../MutableRememberSuspendingTests.kt | 6 +- .../reactive/MutableRememberTests.kt | 6 +- .../reactive/ReactivitySuspendingTests.kt | 28 +++ .../lightningkite/reactive/ReactivityTests.kt | 196 ++++++++++++++++-- 18 files changed, 433 insertions(+), 181 deletions(-) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyChangeListener.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyChangeListener.kt index 270d046..9d1c692 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyChangeListener.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/DependencyChangeListener.kt @@ -131,12 +131,20 @@ suspend fun Reactive.await(): T { } ?: awaitOnce() } +/** + * Reads the current value, waiting for one if there isn't one yet. + * + * A ready state is taken as-is, with no subscription: readiness means some other listener - or the + * source's own nature, as with a `Signal` - is keeping that value current. Only [ReactiveState.notActive] + * and notReady require listening, and the subscription is released as soon as a value arrives. + */ suspend fun Reactive.awaitOnce(): T { val state = state @Suppress("DEPRECATION") return if (state.ready) state.get() else suspendCancellableCoroutine { - // If it's not ready, we need to wait until it is then never bother with this again. + // Not ready, or nothing is maintaining a value: either way we have to listen, which is + // also what activates a lazy source so it calculates one. var remover: (() -> Unit)? = null var alreadyRun = false var done = false diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt index 5e9e002..330c3d4 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt @@ -153,13 +153,6 @@ class TypedReactiveContext( var active = false private set - /** - * When true, dependency-registering operators skip [addListener] and [registerDependency]. - * Set during [runOnceWhileDead] so that a one-off dead read does not subscribe the context - * to any sources, preventing listener leaks and spurious reactivation. - */ - private var skipDependencyRegistration = false - /** * Reference to [startCalculation] used as a listener callback. * Dependencies invoke this when they change to trigger recalculation. @@ -245,27 +238,6 @@ class TypedReactiveContext( } } - /** - * Runs the calculation once without activating the context or tracking dependencies. - * - * The result is reported to [reportTo], but no listeners are registered on any source - * and the calculation will not rerun when dependencies change. - * - * [skipDependencyRegistration] is set to true for the duration so that all - * dependency-registering operators (invoke, async, etc.) skip addListener/registerDependency. - * This prevents the context from leaking listeners onto sources and avoids accidentally - * activating lazy upstream Remembers. - */ - fun runOnceWhileDead() { - skipDependencyRegistration = true - try { - val state = reactiveState { action(this) } - if (!useLastWhileLoading || state.ready) reportTo.state = state - } finally { - skipDependencyRegistration = false - } - } - init { // Automatically cancel when parent scope is cancelled scope.onRemove { cancel() } @@ -316,7 +288,6 @@ class TypedReactiveContext( * Starts using this [ResourceUse] and tracks it as a dependency in future loops. * */ fun use(resourceUse: ResourceUse) { - if (skipDependencyRegistration) return if (existingDependency(resourceUse) != null) return registerDependency(resourceUse, resourceUse.beginUse()) } @@ -336,7 +307,6 @@ class TypedReactiveContext( * ``` */ fun rerunOn(listenable: Listenable) { - if (skipDependencyRegistration) return if (existingDependency(listenable) != null) return registerDependency(listenable, listenable.addListener(rerun)) } @@ -346,6 +316,11 @@ class TypedReactiveContext( * if the state is not ready. * * This allows the reactive calculation to be paused until dependencies become ready. + * + * A notActive dependency counts as loading rather than as an error: every operator here + * registers its listener before reading, so the dependency is being maintained by the time we + * see it, and any notActive we do observe is a source part-way through activating. The + * listener is in place, so the calculation reruns when its value arrives. */ private fun ReactiveState.getOrLoading() = handle( @@ -372,7 +347,7 @@ class TypedReactiveContext( * @throws ReactiveLoading if the value is not ready */ operator fun Reactive.invoke(): R { - if (!skipDependencyRegistration && existingDependency(this) == null) { + if (existingDependency(this) == null) { registerDependency(this, addListener(rerun)) } return state.getOrLoading() @@ -396,7 +371,7 @@ class TypedReactiveContext( * @throws ReactiveLoading if the value is null or not ready */ fun Reactive.awaitNotNull(): R { - if (!skipDependencyRegistration && existingDependency(this) == null) { + if (existingDependency(this) == null) { registerDependency(this, addListener(rerun)) } return state.getOrLoading() ?: throw ReactiveLoading @@ -422,7 +397,7 @@ class TypedReactiveContext( * @return The current [ReactiveState] */ fun Reactive.state(): ReactiveState { - if (!skipDependencyRegistration && existingDependency(this) == null) { + if (existingDependency(this) == null) { registerDependency(this, addListener(rerun)) } return state @@ -447,7 +422,7 @@ class TypedReactiveContext( */ fun Reactive.state(get: (ReactiveState) -> V): V { var current: V = state.let(get) - if (!skipDependencyRegistration && existingDependency(this) == null) { + if (existingDependency(this) == null) { registerDependency(this, addListener { state.let(get) .takeUnless { it == current } @@ -456,6 +431,9 @@ class TypedReactiveContext( rerun() } }) + // Repull in case of activation: a lazy source calculates inside addListener, before + // our listener is in place, so the value read above can predate that calculation. + current = state.let(get) } return current } @@ -464,11 +442,13 @@ class TypedReactiveContext( * Wrapper class for tracking "once" dependencies. * This creates a unique key for each reactive value accessed via [once]. */ - private data class Once(val wraps: Reactive) + private data class Once(val wraps: Reactive) { + /** Whether a value has been obtained; changes stop causing reruns once one has. */ + var have = false + } /** - * Accesses the value of this [Reactive] once, automatically removing the dependency - * after the value becomes ready. + * Accesses the value of this [Reactive] once, without rerunning when it later changes. * * This is useful when you want to wait for an initial value but don't want subsequent * changes to trigger reruns. @@ -485,25 +465,12 @@ class TypedReactiveContext( * @throws ReactiveLoading if the value is not ready yet */ fun Reactive.once(): T { - return state.handle( - success = { it }, - exception = { throw it }, - notReady = { - if (!skipDependencyRegistration) { - val key = Once(this) - if (existingDependency(key) == null) { - // Register a one-shot listener that removes itself after firing - var remover: () -> Unit = {} - remover = addListener { - remover() // Remove the listener - rerun() - } - registerDependency(key, remover) - } - } - throw ReactiveLoading - } - ) + val key = existingDependency(Once(this)) ?: Once(this).also { key -> + registerDependency(key, addListener { if (!key.have) rerun() }) + } + val state = state + key.have = state.ready + return state.getOrLoading() } // Hack: fixes compiler weirdness around lambdas with 'this' @@ -572,24 +539,22 @@ class TypedReactiveContext( val calc = SuspendCalculation(key) // Reuse existing calculation if already running - if (!skipDependencyRegistration) { - existingDependency(calc)?.let { - return it.state.getOrLoading() - } + existingDependency(calc)?.let { + return it.state.getOrLoading() + } - // Launch new calculation and register as dependency. The launch happens before the - // listener is attached so a synchronously-completing action (e.g. under an Unconfined - // dispatcher) can't re-enter rerun() while this calculation is still in progress. - val job = scope.launch { - calc.state = reactiveState { action() } - } - val removeListener = calc.addListener(rerun) - registerDependency(calc) { - // A rerun means this calculation is no longer used - cancel it so stale work - // doesn't keep running and overwrite `calc` after it's been orphaned. - removeListener() - job.cancel() - } + // Launch new calculation and register as dependency. The launch happens before the + // listener is attached so a synchronously-completing action (e.g. under an Unconfined + // dispatcher) can't re-enter rerun() while this calculation is still in progress. + val job = scope.launch { + calc.state = reactiveState { action() } + } + val removeListener = calc.addListener(rerun) + registerDependency(calc) { + // A rerun means this calculation is no longer used - cancel it so stale work + // doesn't keep running and overwrite `calc` after it's been orphaned. + removeListener() + job.cancel() } return calc.state.getOrLoading() @@ -615,23 +580,21 @@ class TypedReactiveContext( operator fun Deferred.invoke(): T { val calc = SuspendCalculation(this) - if (!skipDependencyRegistration) { - // Reuse existing calculation if already running - existingDependency(calc)?.let { - return it.invoke() - } + // Reuse existing calculation if already running + existingDependency(calc)?.let { + return it.invoke() + } - // Launch await operation and register as dependency. Same ordering and cancellation - // rationale as `async` above: launch before attaching the listener, and cancel the - // launch when this dependency is no longer used. - val job = scope.launch { - calc.state = reactiveState { this@invoke.await() } - } - val removeListener = calc.addListener(rerun) - registerDependency(calc) { - removeListener() - job.cancel() - } + // Launch await operation and register as dependency. Same ordering and cancellation + // rationale as `async` above: launch before attaching the listener, and cancel the + // launch when this dependency is no longer used. + val job = scope.launch { + calc.state = reactiveState { this@invoke.await() } + } + val removeListener = calc.addListener(rerun) + registerDependency(calc) { + removeListener() + job.cancel() } return calc.state.getOrLoading() @@ -672,12 +635,6 @@ class TypedReactiveContext( * @throws ReactiveLoading if no value has been emitted yet (except for StateFlow) */ operator fun Flow.invoke(): T { - if (skipDependencyRegistration) { - // Dead read: return current value for StateFlow or notReady for cold flows - if (this is StateFlow) return this.value - else throw ReactiveLoading - } - val new = FlowLoader(this) val existing = existingDependency(new) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt index 8229516..a6e42ab 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContextSuspending.kt @@ -191,33 +191,6 @@ class ReactiveContextSuspending( } } - /** - * Runs the calculation once without activating the context or tracking dependencies. - * - * This is useful for getting an initial value or running the calculation in a test - * environment without setting up the full reactive machinery. - * - * The result is still reported to [reportTo], but no dependency tracking occurs and - * the calculation will not rerun when dependencies change. - */ - fun runOnceWhileDead() { - lastLoopJob = run { - var done = false - val job = scope.launchWithStart { - val result = reactiveState { this@ReactiveContextSuspending.action() } - if (!useLastWhileLoading || result.ready) reportTo.state = result - done = true - } - - // Check if calculation completed synchronously - if (done) null - else { - if (!useLastWhileLoading) reportTo.state = ReactiveState.notReady - job - } - } - } - /** * Called by the dependency tracker when a dependency changes. * Triggers a recalculation with [startCalculation]. diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt index c541a2c..7d4986b 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt @@ -14,7 +14,7 @@ import kotlin.time.Duration * is manually set it will stop the calculation and instead behave like a [Signal]. * * Note: - * - `mutableRemember` is lazy: if it has no listeners, it will not calculate a value. + * - `mutableRemember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none, and has not been set, reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated or set value changes. * * Example: @@ -46,7 +46,7 @@ fun mutableRemember( * When overridden by direct assignment, automatic calculation is paused until `reset()` is called. * * Note: - * - `MutableRemember` is lazy: if it has no listeners, it will not calculate a value. + * - `MutableRemember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none, and has not been set, reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated or set value changes. * - The `reset()` method restores automatic calculation and updates the value from dependencies. * @@ -70,16 +70,12 @@ class MutableRemember( private val remember = Remember(coroutineContext, useLastWhileLoading, deactivationDelay, initialValue) private var forget: (()->Unit)? = null - private fun updateOnce() { - val currentRememberedState = remember.state - if(!overridden && (!useLastWhileLoading || currentRememberedState.ready)) state = currentRememberedState - } - private fun startListening() { forget = remember.addListener { if (!overridden) state = remember.state } - updateOnce() + val currentRememberedState = remember.state + if (!overridden && (!useLastWhileLoading || currentRememberedState.ready)) state = currentRememberedState } private fun stopListening() { forget?.invoke() @@ -88,7 +84,7 @@ class MutableRemember( override var state: ReactiveState get() { - if (!overridden && forget == null) updateOnce() + if (!overridden && forget == null) return ReactiveState.notActive return super.state } set(value) { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRememberSuspending.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRememberSuspending.kt index 15f0629..d837487 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRememberSuspending.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRememberSuspending.kt @@ -14,7 +14,7 @@ import kotlin.time.Duration * is manually set it will stop the calculation and instead behave like a [Signal]. * * Note: - * - `mutableRememberSuspending` is lazy: if it has no listeners, it will not calculate a value. + * - `mutableRememberSuspending` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none, and has not been set, reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated or set value changes. * * Example: @@ -47,7 +47,7 @@ fun mutableRememberSuspending( * When overridden by direct assignment, automatic calculation is paused until `reset()` is called. * * Note: - * - lazy: if this has no listeners, it will not calculate a value. + * - lazy: if this has no listeners, it will not calculate a value. Reading its state while it has none, and has not been set, reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated or set value changes. I.e., if '1' is calculated, and then '1' is set, it will not notify listeners. * - The `reset()` method restores automatic calculation and updates the value from dependencies. * @@ -90,7 +90,7 @@ class MutableRememberSuspending( override var state: ReactiveState get() { - if (!overridden && forget == null) updateOnce() + if (!overridden && forget == null) return ReactiveState.notActive return super.state } set(value) { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt index 74d69a3..88d9986 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt @@ -7,16 +7,34 @@ import kotlin.jvm.JvmInline /** * Represents the state of a reactive value, including loading, success, and error conditions. * - * - A [ReactiveState] can hold a ready value, a loading state, or an error state. + * - A [ReactiveState] can hold a ready value, a loading state, an error state, or [notActive]. * - Use [ready] to check if the value is available, [success] to check if it is available and not an error, and [exception] to retrieve any error. * - Listeners of [Reactive] are only notified when the [ReactiveState] changes. * - [ReactiveState] provides methods for safely handling, mapping, and retrieving the underlying value. + * + * ### notReady versus notActive + * + * Both mean "no value", for different reasons, and both make [ready] false: + * + * - [Companion.notReady]: something is maintaining this value, and it does not have one right now - + * it is loading, or one of its dependencies is not ready. + * - [Companion.notActive]: *nothing* is maintaining this value. Lazy reactives like `remember` only + * calculate while they have listeners, so with none they cannot vouch for any value. Sources whose + * value is accurate whether or not anyone is listening - `Signal`, `Constant` - never report it. + * + * The distinction is what makes it safe to read a value without subscribing: anything other than + * [notActive] is current, no matter who is (or isn't) listening. On [notActive] you must subscribe - + * see `awaitOnce` - or accept that there is no value to be had. */ @JvmInline @OptIn(InternalReactiveApi::class) value class ReactiveState(val raw: T) { - inline val ready: Boolean get() = raw !is InternalReactiveNotReady + inline val ready: Boolean get() = raw !is InternalReactiveNotReady && raw !is InternalReactiveNotActive inline val success: Boolean get() = ready && raw !is InternalReactiveThrownException + + /** True when nothing is maintaining this value; see the [ReactiveState] docs. */ + inline val notActive: Boolean get() = raw is InternalReactiveNotActive + inline fun onSuccess(action: (T)->R): R? = handle( success = { action(it) }, exception = { null }, @@ -28,7 +46,8 @@ value class ReactiveState(val raw: T) { fun get(): T = handle( success = { it }, exception = { throw it }, - notReady = { throw NotReadyException() } + notReady = { throw NotReadyException() }, + notActive = { throw NotActiveException() } ) fun getOrNull(): T? = handle( @@ -40,6 +59,11 @@ value class ReactiveState(val raw: T) { companion object Companion { @Suppress("UNCHECKED_CAST") val notReady: ReactiveState = ReactiveState(InternalReactiveNotReady) as ReactiveState + + /** No value, because nothing is maintaining one; see the [ReactiveState] docs. */ + @Suppress("UNCHECKED_CAST") + val notActive: ReactiveState = ReactiveState(InternalReactiveNotActive) as ReactiveState + @Suppress("UNCHECKED_CAST") fun exception(exception: Exception) = (if(exception is CancellationException) notReady else ReactiveState(InternalReactiveThrownException(exception))) as ReactiveState @Suppress("UNCHECKED_CAST") @@ -47,7 +71,9 @@ value class ReactiveState(val raw: T) { } @Suppress("UNCHECKED_CAST") inline fun map(mapper: (T)->B): ReactiveState { - if(raw is InternalReactiveNotReady || raw is InternalReactiveThrownException) return this as ReactiveState + // notActive propagates like the other valueless states: a value derived from a source + // nobody is maintaining is equally unmaintained. + if(raw is InternalReactiveNotReady || raw is InternalReactiveNotActive || raw is InternalReactiveThrownException) return this as ReactiveState if(raw is InternalReactiveWrapper<*>) try { return ReactiveState(mapper(raw.other as T)) } catch(e: Exception) { @@ -59,24 +85,43 @@ value class ReactiveState(val raw: T) { exception(e) } } - @Suppress("UNCHECKED_CAST") + /** + * Handles [Companion.notActive] the same way as [notReady], because "nobody is maintaining a + * value" and "there is no value yet" are the same thing to code that only wants to display or + * wait for one. Use the four-argument overload to do something better, such as subscribing. + */ inline fun handle( success: (T)->R, exception: (Exception)->R, notReady: ()->R + ): R = handle(success, exception, notReady, notReady) + + @Suppress("UNCHECKED_CAST") + inline fun handle( + success: (T)->R, + exception: (Exception)->R, + notReady: ()->R, + notActive: ()->R ): R { return when(raw) { InternalReactiveNotReady -> notReady() + InternalReactiveNotActive -> notActive() is InternalReactiveThrownException -> exception(raw.exception) is InternalReactiveWrapper<*> -> success(raw.other as T) else -> success(raw) } } - fun asResult(): Result = handle(success = { Result.success(it) }, exception = { Result.failure(it) }, notReady = { Result.failure(NotReadyException()) }) + fun asResult(): Result = handle( + success = { Result.success(it) }, + exception = { Result.failure(it) }, + notReady = { Result.failure(NotReadyException()) }, + notActive = { Result.failure(NotActiveException()) } + ) override fun toString(): String = when(raw) { is InternalReactiveNotReady -> "NotReady" + is InternalReactiveNotActive -> "NotActive" is InternalReactiveThrownException -> "ThrownException(${raw.exception})" is InternalReactiveWrapper<*> -> "ReadyW($raw)" else -> "Ready($raw)" @@ -88,8 +133,17 @@ data class InternalReactiveWrapper(val other: T) data class InternalReactiveThrownException(val exception: Exception) @InternalReactiveApi object InternalReactiveNotReady +@InternalReactiveApi +object InternalReactiveNotActive + +open class NotReadyException(message: String? = null) : IllegalStateException(message) -class NotReadyException(message: String? = null) : IllegalStateException(message) +/** + * Thrown when reading a value from a reactive that nothing is listening to, and which therefore + * has no value to give. Subscribe to it first, or use `awaitOnce`, which subscribes for as long + * as it takes to obtain a value. + */ +class NotActiveException(message: String = "Nothing is listening to this reactive value, so it has no value to report. Subscribe to it, or use awaitOnce.") : NotReadyException(message) inline fun reactiveState(action: () -> T): ReactiveState { @OptIn(InternalReactiveApi::class) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt index ecbe95a..0fee54b 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt @@ -14,7 +14,7 @@ import kotlin.time.Duration * The calculation runs in a coroutine, and will re-run whenever any reactive value it depends on changes. * * Note: - * - `remember` is lazy: if it has no listeners, it will not calculate a value. + * - `remember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated value changes (i.e., if the new value is different from the previous value). * * @param coroutineContext The coroutine context for running the calculation (default: Dispatchers.Unconfined). @@ -54,7 +54,7 @@ fun remember( * any of its dependencies change, and listeners are notified accordingly. * * Note: - * - `Remember` is lazy: if it has no listeners, it will not calculate a value. + * - `Remember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated value changes (i.e., if the new value is different from the previous value). * * @param T The type of value produced by the calculation. @@ -71,7 +71,7 @@ fun remember( */ class Remember( val incomingCoroutineContext: CoroutineContext = Dispatchers.Unconfined, - useLastWhileLoading: Boolean = false, + private val useLastWhileLoading: Boolean = false, private val deactivationDelay: Duration? = null, private val action: ReactiveContext.() -> T, ) : Reactive, CoroutineScope, BaseListenable() { @@ -92,13 +92,17 @@ class Remember( // Remember/shared reactive graph forever. Matches RememberSuspending's ordering. override val coroutineContext get() = restOfContext + job - private val scope = TypedReactiveContext(this, useLastWhileLoading, action = action) + // Starts notActive rather than notReady: nothing is listening yet, so there is no value to + // be had, as opposed to one that is on its way. + private val reported = RawReactive(ReactiveState.notActive) + private val scope = TypedReactiveContext(this, useLastWhileLoading, reported, action) - override val state: ReactiveState - get() { - if (!scope.active) scope.runOnceWhileDead() - return scope.state - } + // A Remember only calculates while it has listeners, and reports notActive when it has none. + // It deliberately does not calculate on demand: doing so would either subscribe to sources + // nobody is listening to, or run a dependency-less calculation whose result is silently never + // updated. To read one imperatively, use awaitOnce - it subscribes for as long as it takes to + // get a value. + override val state: ReactiveState get() = reported.state private var deactivating: Job? = null private var remover: (() -> Unit)? = null @@ -111,6 +115,11 @@ class Remember( return } + // Something is maintaining this value again - it just doesn't have one yet. Without this, + // useLastWhileLoading would suppress the notReady the first calculation reports and leave + // notActive in place, claiming nobody is listening when somebody now is. + if (reported.state.notActive) reported.state = ReactiveState.notReady + shuttingDown?.let { CoroutineScope(incomingCoroutineContext).launch { it.join() @@ -130,6 +139,9 @@ class Remember( job.cancel() job = SupervisorJob() shuttingDown = null + // The calculation is stopped, so the value it produced is no longer maintained. Reporting + // notActive says exactly that, rather than passing off a value that may have gone stale. + reported.state = ReactiveState.notActive } override fun deactivate() { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/RememberSuspending.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/RememberSuspending.kt index 04a7220..0d0e43b 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/RememberSuspending.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/RememberSuspending.kt @@ -14,7 +14,7 @@ import kotlin.time.Duration * The calculation runs in a coroutine, and will re-run whenever any reactive value it depends on changes. * * Note: - * - `remember` is lazy: if it has no listeners, it will not calculate a value. + * - `remember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated value changes (i.e., if the new value is different from the previous value). * * @param coroutineContext The coroutine context for running the calculation (default: Dispatchers.Unconfined). @@ -56,7 +56,7 @@ fun rememberSuspending( * any of its dependencies change, and listeners are notified accordingly. * * Note: - * - `Remember` is lazy: if it has no listeners, it will not calculate a value. + * - `Remember` is lazy: if it has no listeners, it will not calculate a value. Reading its state while it has none reports [ReactiveState.Companion.notActive]. * - Listeners are only notified if the calculated value changes (i.e., if the new value is different from the previous value). * * @param T The type of value produced by the calculation. @@ -74,7 +74,7 @@ fun rememberSuspending( */ class RememberSuspending( val incomingCoroutineContext: CoroutineContext = Dispatchers.Unconfined, - useLastWhileLoading: Boolean = false, + private val useLastWhileLoading: Boolean = false, private val deactivationDelay: Duration? = null, private val action: suspend ReactiveCoroutineScope.() -> T, ) : Reactive, CoroutineScope, BaseListenable() { @@ -89,13 +89,12 @@ class RememberSuspending( override val coroutineContext: CoroutineContext get() = restOfContext + job - private val scope = ReactiveContextSuspending(this, useLastWhileLoading, action = action) + // See [Remember.reported]: nothing is listening yet, so there is no value to be had. + private val reported = RawReactive(ReactiveState.notActive) + private val scope = ReactiveContextSuspending(this, useLastWhileLoading, reported, action) - override val state: ReactiveState - get() { - if (!scope.active) scope.runOnceWhileDead() - return scope.state - } + // See [Remember.state]: no listeners means no calculation, and so notActive. + override val state: ReactiveState get() = reported.state private var deactivating: Job? = null private var remover: (() -> Unit)? = null @@ -108,6 +107,9 @@ class RememberSuspending( return } + // See [Remember.activate]: listening again means active-but-valueless, not notActive. + if (reported.state.notActive) reported.state = ReactiveState.notReady + shuttingDown?.let { CoroutineScope(incomingCoroutineContext).launch { it.join() @@ -127,6 +129,8 @@ class RememberSuspending( job.cancel() job = SupervisorJob() shuttingDown = null + // See [Remember.shutdown]: a stopped calculation maintains no value. + reported.state = ReactiveState.notActive } override fun deactivate() { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/coreInterfaces.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/coreInterfaces.kt index ed1c2e5..2af73fc 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/coreInterfaces.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/coreInterfaces.kt @@ -92,6 +92,14 @@ fun Listenable.addAndRunListener(listener: () -> Unit): Release { * @see com.lightningkite.reactive.context.ReactiveContext */ interface Reactive : Listenable { + /** + * The current state. + * + * Implementations that only maintain a value while they have listeners - `remember` and + * friends - must report [ReactiveState.Companion.notActive] when they have none, rather than + * a value they are no longer keeping up to date. Anything else is a promise that the state is + * current, which is what lets callers read it without subscribing. + */ val state: ReactiveState object Never: Reactive { @@ -196,6 +204,9 @@ interface MutableReactive : Reactive, Mutable { */ interface ReactiveValue : Reactive, ReadOnlyProperty { val value: T + + // A ReactiveValue always has a value, so its state can never be notReady or notActive. That + // makes it the wrong interface for anything that only maintains a value while listened to. override val state: ReactiveState get() = ReactiveState(value) override fun getValue(thisRef: Any?, property: KProperty<*>): T = value } diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/processes.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/processes.kt index 575cd39..4891cef 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/processes.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/processes.kt @@ -23,8 +23,10 @@ fun CoroutineScope.reactiveProcess(emitter: suspend Emitter.() -> Unit): } return prop } +// The two below only run their emitter while something is listening, so with no listeners they +// report notActive: whatever the emitter last produced is no longer being kept up to date. fun reactiveProcess(scope: CoroutineScope = AppScope, emitter: suspend Emitter.() -> Unit): Reactive { - return object: BaseReactive() { + return object: BaseReactive(ReactiveState.notActive) { var job: Job? = null override fun activate() { state = ReactiveState.notReady @@ -39,13 +41,15 @@ fun reactiveProcess(scope: CoroutineScope = AppScope, emitter: suspend Emitt override fun deactivate() { job?.cancel() job = null + state = ReactiveState.notActive } } } fun rawReactiveProcess(scope: CoroutineScope = AppScope, emitter: suspend Emitter>.() -> Unit): Reactive { - return object: BaseReactive() { + return object: BaseReactive(ReactiveState.notActive) { var job: Job? = null override fun activate() { + state = ReactiveState.notReady job = scope.launch { emitter(object : Emitter>, CoroutineScope by this@launch { override fun emit(value: ReactiveState) { @@ -57,6 +61,7 @@ fun rawReactiveProcess(scope: CoroutineScope = AppScope, emitter: suspend Em override fun deactivate() { job?.cancel() job = null + state = ReactiveState.notActive } } } \ No newline at end of file diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/extensions/WaitForNotNull.kt b/src/commonMain/kotlin/com/lightningkite/reactive/extensions/WaitForNotNull.kt index 7e51279..bbb098f 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/extensions/WaitForNotNull.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/extensions/WaitForNotNull.kt @@ -8,6 +8,8 @@ import com.lightningkite.reactive.core.Release import kotlinx.coroutines.suspendCancellableCoroutine internal class WaitForNotNull(val wraps: Reactive) : Reactive { + // A null value is notReady - a value is there, it just isn't one we can use yet. Every other + // state, notActive included, passes through as the wrapped reactive reported it. @Suppress("UNCHECKED_CAST") override val state: ReactiveState get() = if(wraps.state.raw == null) ReactiveState.notReady else wraps.state as ReactiveState diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt b/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt index d821f12..3faca02 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt @@ -61,18 +61,31 @@ fun Reactive.withWrite(action: suspend Reactive.(T) -> Unit): MutableR } fun Reactive.onNextSuccess(action: (T) -> Unit): Release? { + // A successful state is one somebody is maintaining, so it needs no subscription at all. if (state.success) { state.onSuccess(action) return null } var release: Release? = null + var acted = false release = addListener { state.onSuccess { + acted = true action(it) release?.invoke() } } + // Read after subscribing: activating a notActive source calculates inside addListener, before + // our listener is in place. + state.onSuccess { + acted = true + action(it) + } + if (acted) { + release?.invoke() + return null + } return release } @@ -130,7 +143,9 @@ fun Reactive>.flatten(): Reactive = remember { this@flatten() fun Reactive>.flatten(): MutableReactive = remember { this@flatten()() }.withWrite { - this@flatten.state.onSuccess { s -> s set it } + // awaitOnce rather than reading state: if the outer reactive is lazy it has no value to + // read unless something is listening, and the write would be silently dropped. + this@flatten.awaitOnce().set(it) } fun CoroutineScope.asyncReactive(action: suspend () -> T): Reactive { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/lensing/abstracts.kt b/src/commonMain/kotlin/com/lightningkite/reactive/lensing/abstracts.kt index d19db49..b489bae 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/lensing/abstracts.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/lensing/abstracts.kt @@ -22,12 +22,15 @@ open class Lens, T, L>(val source: S, val get: (T) -> L) : BaseR private var myListen: (() -> Unit)? = null + // Subscribe before reading: activating a lazy source (a Remember, for example) calculates its + // state as part of addListener, and that calculation happens before our listener is in place. + // Reading afterwards is what picks that first value up. override fun activate() { super.activate() - super.state = source.state.map(get) myListen = source.addListener { super.state = source.state.map(get) } + super.state = source.state.map(get) } override fun deactivate() { @@ -74,12 +77,14 @@ open class ValueLens, T, L>( } private var myListen: (() -> Unit)? = null + + // Subscribe before reading, for the same reason as [Lens.activate]. override fun activate() { super.activate() - super.value = source.value.let(get) myListen = source.addListener { super.value = source.value.let(get) } + super.value = source.value.let(get) } override fun deactivate() { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/lensing/validation/validation.kt b/src/commonMain/kotlin/com/lightningkite/reactive/lensing/validation/validation.kt index 41ad118..6f84f7e 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/lensing/validation/validation.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/lensing/validation/validation.kt @@ -44,10 +44,11 @@ private open class ValidatedLens, T>( override fun activate() { super.activate() baseNode.connect() - state = source.state.also(::check) + // Subscribe before reading, so state produced by activating a lazy source isn't missed. myListen = source.addListener { if (listen) state = source.state.also(::check) } + state = source.state.also(::check) } override fun deactivate() { super.deactivate() @@ -88,10 +89,11 @@ private open class ValidatedValueLens, T>( override fun activate() { super.activate() baseNode.connect() - value = source.value.also(::check) + // Subscribe before reading, for the same reason as [ValidatedLens.activate]. myListen = source.addListener { if (listen) value = source.value.also(::check) } + value = source.value.also(::check) } override fun deactivate() { super.deactivate() diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberSuspendingTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberSuspendingTests.kt index c5071f1..2eaecef 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberSuspendingTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberSuspendingTests.kt @@ -231,8 +231,12 @@ class MutableRememberSuspendingTests { late.unset() assertEquals(ReactiveState(10), test.state) + // Resetting restarts the calculation, which has nothing to calculate from now that + // `late` is unset. useLastWhileLoading means the last value stands until it does. + // (It used to report 1 here - a value the dormant calculation had held onto from + // before `late` was unset.) test.reset() - assertEquals(ReactiveState(1), test.state) + assertEquals(ReactiveState(10), test.state) late.value = 2 assertEquals(ReactiveState(2), test.state) diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberTests.kt index d83c19e..7fee299 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/MutableRememberTests.kt @@ -228,8 +228,12 @@ class MutableRememberTests { late.unset() assertEquals(ReactiveState(10), test.state) + // Resetting restarts the calculation, which has nothing to calculate from now that + // `late` is unset. useLastWhileLoading means the last value stands until it does. + // (It used to report 1 here - a value the dormant calculation had held onto from + // before `late` was unset.) test.reset() - assertEquals(ReactiveState(1), test.state) + assertEquals(ReactiveState(10), test.state) late.value = 2 assertEquals(ReactiveState(2), test.state) diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt index afb4307..580da49 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivitySuspendingTests.kt @@ -28,6 +28,7 @@ import kotlinx.coroutines.test.runTest import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertFailsWith +import kotlin.test.assertFalse import kotlin.test.assertIs import kotlin.test.assertTrue @@ -494,4 +495,31 @@ class ReactivitySuspendingTests { assertEquals(5, innerCancelled) } } + + @Test + fun readingStateWithoutListenersDoesNotCalculate() { + val source = LateInitSignal() + var computeCount = 0 + val r = rememberSuspending { + computeCount++ + source() + } + + assertEquals(0, source.listenerCount, "precondition: no listeners before reading state") + + // Reading state repeatedly must neither calculate nor subscribe: a calculation that waits + // on a not-ready source would hold a subscription open for as long as it waits. + assertEquals(ReactiveState.notActive, r.state) + assertEquals(ReactiveState.notActive, r.state) + assertEquals(0, computeCount, "reading state must not calculate") + assertEquals(0, source.listenerCount, "reading state must not subscribe to sources") + + val release = r.addListener { } + assertEquals(1, source.listenerCount, "activating subscribes") + source.value = 3 + assertEquals(3, r.state.get()) + release() + assertEquals(0, source.listenerCount, "deactivating releases the subscription") + assertEquals(ReactiveState.notActive, r.state, "a dormant Remember must not report its last value") + } } diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index f064d12..ddeea65 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -10,11 +10,17 @@ import com.lightningkite.reactive.context.ReactiveReentrancyException import com.lightningkite.reactive.core.Reactive import com.lightningkite.reactive.core.ReactiveState import com.lightningkite.reactive.core.addAndRunListener +import com.lightningkite.reactive.context.awaitOnce +import com.lightningkite.reactive.extensions.flatten import com.lightningkite.reactive.extensions.interceptWrite +import com.lightningkite.reactive.extensions.onNextSuccess +import com.lightningkite.reactive.lensing.lens import com.lightningkite.reactive.extensions.value import com.lightningkite.reactive.extensions.waitForNotNull import com.lightningkite.reactive.core.LateInitSignal import com.lightningkite.reactive.core.RawReactive +import com.lightningkite.reactive.core.MutableReactive +import com.lightningkite.reactive.core.NotActiveException import com.lightningkite.reactive.core.Remember import com.lightningkite.reactive.core.Signal import com.lightningkite.reactive.core.remember @@ -593,7 +599,7 @@ class ReactivityTests { } @Test - fun readingDeadStateDoesNotLeakOrResurrect() { + fun readingStateWithoutListenersDoesNotCalculate() { val source = Signal(1) var computeCount = 0 val r = remember { @@ -601,21 +607,187 @@ class ReactivityTests { source() } - // No listener added, so the Remember is lazy/dead. - assertEquals(0, source.listenerCount, "precondition: no listeners before touching dead state") + // No listener added, so the Remember is lazy and has never calculated. + assertEquals(0, source.listenerCount, "precondition: no listeners before reading state") - // Reading .state on a dead Remember computes a one-off value. - assertEquals(1, r.state.get(), "dead read should still compute the current value") - val computesAfterRead = computeCount - assertTrue(computesAfterRead > 0, "dead read should have computed at least once") + assertEquals(ReactiveState.notActive, r.state, "a Remember with no listeners maintains no value") + assertEquals(0, computeCount, "reading state must not calculate") + assertEquals(0, source.listenerCount, "reading state must not subscribe to sources") - // The throwaway computation must not leave a listener on the source. - assertEquals(0, source.listenerCount, "reading dead .state must not leave a dangling listener on the source") - - // Mutating the source must NOT resurrect the Remember. + // Mutating the source must not resurrect it either. source.value = 2 - assertEquals(computesAfterRead, computeCount, "mutating the source must not recompute a dead Remember") + assertEquals(0, computeCount, "mutating a source must not calculate a Remember nobody listens to") assertEquals(0, source.listenerCount, "source must still have zero listeners after mutation") + + // Gaining a listener is what makes it calculate, and losing it stops it again. + val release = r.addListener { } + assertEquals(2, r.state.get(), "activating calculates the current value") + assertEquals(1, computeCount) + assertEquals(1, source.listenerCount) + release() + assertEquals(0, source.listenerCount, "deactivating releases the subscription") + + // Going dormant drops the value rather than reporting one that is no longer maintained. + assertEquals(ReactiveState.notActive, r.state, "a dormant Remember must not report its last value") + source.value = 3 + assertEquals(1, computeCount, "a dormant Remember must not recalculate") + + // Listening again recalculates against the source as it is now. + r.addListener { } + assertEquals(3, r.state.get(), "reactivating recalculates rather than reporting the old value") + } + + // The tests below cover reading a lazy source: because a Remember only calculates while + // something is listening, and BaseListenable activates before the new listener is in the list, + // every one of these has to subscribe before reading or it sees the pre-activation state. + + @Test + fun lensOfLazySourceGetsValueOnActivation() { + val source = Signal(1) + val lensed = remember { source() }.lens { it * 10 } + val release = lensed.addListener { } + assertEquals(10, lensed.state.get(), "activating a lens must pick up the value its source calculates") + source.value = 2 + assertEquals(20, lensed.state.get()) + release() + } + + @Test + fun onceObtainsValueFromLazySource() { + val source = Signal(1) + val r = remember { source() } + val seen = ArrayList() + testContext { + reactive { seen.add(r.once()) } + assertEquals(listOf(1), seen, "once must obtain a value from a lazy source") + source.value = 2 + source.value = 3 + assertEquals(listOf(1), seen, "once must not rerun the calculation, nor change what it reports") + } + } + + @Test + fun stateTransformSeesValueFromActivation() { + val source = Signal(1) + val r = remember { source() } + val seen = ArrayList() + testContext { + reactive { seen.add(r.state { it.ready }) } + assertEquals(listOf(true), seen, "state(get) must see the value its subscription caused") + source.value = 2 + assertEquals(listOf(true), seen, "readiness did not change, so there is nothing to rerun") + } + } + + @Test + fun onNextSuccessFiresForLazySource() { + val source = Signal(1) + val seen = ArrayList() + remember { source() }.onNextSuccess { seen.add(it) } + assertEquals(listOf(1), seen, "onNextSuccess must obtain a value from a lazy source") + source.value = 2 + assertEquals(listOf(1), seen, "onNextSuccess fires exactly once") + } + + @Test + fun writingThroughFlattenedLazySourceReachesTheTarget() = runTest { + val inner = Signal(1) + val selection = Signal>(inner) + // The outer reactive is lazy, so the write has to activate it to find the target. + val outer: Reactive> = remember { selection() } + outer.flatten().set(42) + assertEquals(42, inner.value, "a write through flatten must not be silently dropped") + } + + @Test + fun notActiveIsDistinctFromLoading() { + val source = LateInitSignal() + val r = remember { source() } + + assertEquals(ReactiveState.notActive, r.state, "nobody is maintaining this") + + val release = r.addListener { } + assertEquals( + ReactiveState.notReady, r.state, + "now somebody is maintaining it - it just doesn't have a value yet" + ) + + source.value = 1 + assertEquals(ReactiveState(1), r.state) + + release() + assertEquals(ReactiveState.notActive, r.state, "and back to nobody maintaining it") + } + + @Test + fun notActivePropagatesThroughLenses() { + val source = Signal(1) + val r = remember { source() } + val lensed = r.lens { it * 10 } + + assertTrue(lensed.state.notActive, "a lens over an unmaintained value is equally unmaintained") + + val release = lensed.addListener { } + assertEquals(ReactiveState(10), lensed.state) + release() + assertTrue(lensed.state.notActive) + } + + @Test + fun lensOfAlwaysDefiniteSourceNeedsNoListeners() { + // A Signal's value is accurate whether or not anyone is listening, so a lens over one can + // be read directly - this is what notActive being a separate state buys. + val source = Signal(1) + val lensed = source.lens { it * 10 } + assertEquals(ReactiveState(10), lensed.state) + source.value = 2 + assertEquals(ReactiveState(20), lensed.state) + } + + @Test + fun readingAValueSomebodyElseMaintainsIsFree() = runTest { + val source = Signal(1) + var computeCount = 0 + val r = remember { + computeCount++ + source() + } + val release = r.addListener { } + assertEquals(1, computeCount) + + // Someone else is keeping this current, so reading it must not re-run anything or churn + // the subscription. + assertEquals(1, r.awaitOnce()) + assertEquals(1, computeCount, "awaitOnce must not recalculate a value already being maintained") + assertEquals(1, source.listenerCount, "awaitOnce must not disturb the existing subscription") + + release() + } + + @Test + fun readingNotActiveStateThrowsSomethingExplanatory() { + val r = remember { Signal(1)() } + @Suppress("DEPRECATION") + val thrown = assertFailsWith { r.state.get() } + assertTrue(thrown.message!!.contains("listening"), "the message should say what to do about it") + } + + @Test + fun awaitOnceCalculatesAndReleases() = runTest { + val source = Signal(1) + var computeCount = 0 + val r = remember { + computeCount++ + source() + } + + assertEquals(1, r.awaitOnce(), "awaitOnce must calculate a dormant Remember") + assertEquals(1, computeCount) + assertEquals(0, source.listenerCount, "awaitOnce must not hold onto its subscription") + + source.value = 2 + assertEquals(2, r.awaitOnce(), "awaitOnce must recalculate rather than report a stale value") + assertEquals(0, source.listenerCount) } } From 8747c55546bdffc586c4e0480e642543461f94e5 Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Wed, 29 Jul 2026 10:41:06 -0600 Subject: [PATCH 10/12] New reentrancy detection stuff --- .../reactive/context/ReactiveContext.kt | 95 ++++++++-------- .../lightningkite/reactive/core/Remember.kt | 2 +- .../lightningkite/reactive/ReactivityTests.kt | 102 ++++++++++++++---- 3 files changed, 128 insertions(+), 71 deletions(-) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt index 330c3d4..57e4032 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt @@ -141,39 +141,18 @@ typealias ReactiveContext = TypedReactiveContext<*> class TypedReactiveContext( val scope: CoroutineScope, val useLastWhileLoading: Boolean = false, + val reentrancyLimit: Int = 0, private val reportTo: RawReactive = RawReactive(), val action: TypedReactiveContext.() -> T ) : DependencyChangeListener(), ReactiveCoroutineScope, Reactive by reportTo { companion object - /** - * Whether this context is currently active and tracking dependencies. - * Set to false when [cancel] is called. - */ - var active = false - private set - /** * Reference to [startCalculation] used as a listener callback. * Dependencies invoke this when they change to trigger recalculation. */ val rerun: () -> Unit = ::startCalculation - /** - * Prevents multiple simultaneous recalculation requests from queueing up. - * Only the first request triggers [startCalculation], subsequent requests are ignored until it completes. - */ - private var queued = false - - /** - * True while this context's [action] is executing. Used to detect reentrancy: if the - * calculation writes to a signal it also reads, the write synchronously re-invokes - * [startCalculation] on this same context, which would otherwise recurse until the stack - * overflows (or livelock under a dispatching scheduler). Detecting it lets us fail fast with - * a clear message instead. - */ - private var calculating = false - /** * The current job for this calculation run. * Gets cancelled and replaced with a new job on each [startCalculation] call. @@ -192,6 +171,10 @@ class TypedReactiveContext( */ override val coroutineContext: CoroutineContext get() = scope.coroutineContext + job + this + private var queued = false + private var desired = false + private var calculating = false + /** * Starts or restarts the reactive calculation. * @@ -206,34 +189,34 @@ class TypedReactiveContext( * Thread safety: Uses [queued] flag to prevent multiple simultaneous executions. */ fun startCalculation() { - // A rerun requested while we're mid-calculation means the calculation triggered its own - // dependency to change (e.g. wrote to a signal it reads). Fail fast rather than recurse - // until the stack overflows. The listener dispatch that called us is synchronous, so this - // check catches the offending write even under a dispatching scheduler. - if (calculating) throw ReactiveReentrancyException(this) - active = true + desired = true if (queued) return // Prevent duplicate queuing queued = true - - // Cancel previous calculation and create fresh job - job.cancel() - job = Job() + if (calculating) return scope.onThread { - queued = false - if (!active) return@onThread // Check if cancelled while queued - - calculating = true - try { - dependencyBlockStart() // Begin tracking dependencies - val state = reactiveState { action(this@TypedReactiveContext) } - - // Update state unless useLastWhileLoading is true and result isn't ready - if (!useLastWhileLoading || state.ready) reportTo.state = state - - dependencyBlockEnd() // Clean up dependencies not used in this run - } finally { - calculating = false + if (!desired) return@onThread // Check if cancelled while queued + + var iter = 0 + while(queued) { + queued = false + job.cancel() + job = Job() + if (iter++ > reentrancyLimit) { + reportTo.state = ReactiveState.exception(ReactiveReentrancyException(this)) + break + } + try { + calculating = true + dependencyBlockStart() // Begin tracking dependencies + val state = reactiveState { action(this@TypedReactiveContext) } + + // Update state unless useLastWhileLoading is true and result isn't ready + if (!useLastWhileLoading || state.ready) reportTo.state = state + } finally { + dependencyBlockEnd() // Clean up dependencies not used in this run + calculating = false + } } } } @@ -273,7 +256,7 @@ class TypedReactiveContext( override fun cancel() { job.cancel() job = Job() - active = false + desired = false queued = false super.cancel() // Cancel dependency listeners } @@ -643,21 +626,29 @@ class TypedReactiveContext( // Register cleanup to cancel collection when dependency is removed registerDependency(new, { job?.cancel() }) - // Start collecting the flow + // An emission that arrives while we are still registering (a hot flow under an + // undispatched scheduler collects synchronously) belongs to this run, so it is + // delivered by the return below rather than through rerun(). A dependency's initial + // value is not the calculation triggering itself, and pushing it through rerun() + // makes it indistinguishable from reentrancy. Same rationale as the + // launch-before-listen ordering in `async` above. + var registering = true job = scope.launch { collect { v -> try { new.state = ReactiveState(v) - rerun() // Trigger recalculation on each emission + if (!registering) rerun() // Later emissions trigger recalculation } catch (e: Exception) { new.state = ReactiveState.exception(e) } } } + registering = false // StateFlow always has a current value available immediately if (this is StateFlow) return this.value - else throw ReactiveLoading + // A flow that emitted during registration already has a value for this run + else return new.state.getOrLoading() } else { return existing.state.handle( success = { it }, @@ -726,8 +717,8 @@ class TypedReactiveContext( * @see TypedReactiveContext for implementation details * @see reactiveSuspending for suspending calculations */ -fun CoroutineScope.reactive(action: ReactiveContext.() -> T): TypedReactiveContext { - val trc = TypedReactiveContext(this, action = action) +fun CoroutineScope.reactive(reentrancyLimit: Int = 0, action: ReactiveContext.() -> T): TypedReactiveContext { + val trc = TypedReactiveContext(this, reentrancyLimit = reentrancyLimit, action = action) trc.startCalculation() coroutineContext[StatusListener]?.watchBackgroundProcess(trc) return trc diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt index 0fee54b..1a0aa1e 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt @@ -95,7 +95,7 @@ class Remember( // Starts notActive rather than notReady: nothing is listening yet, so there is no value to // be had, as opposed to one that is on its way. private val reported = RawReactive(ReactiveState.notActive) - private val scope = TypedReactiveContext(this, useLastWhileLoading, reported, action) + private val scope = TypedReactiveContext(this, useLastWhileLoading, 0, reported, action) // A Remember only calculates while it has listeners, and reports notActive when it has none. // It deliberately does not calculate on demand: doing so would either subscribe to sources diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index ddeea65..34ae9e6 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -504,26 +504,92 @@ class ReactivityTests { @Test fun reentrancyThrowsClearError() { - val previous = Reactive.reportException - val captured = ArrayList() - Reactive.reportException = { captured.add(it) } - try { - testContext { - val s = Signal(0) - // Writing to a signal the calculation also reads re-triggers this same calculation. - // Without reentrancy detection this recurses until the stack overflows. - reactive { - val v = s() - s.value = v + 1 - } + testContext { + val s = Signal(0) + // Writing to a signal the calculation also reads re-triggers this same calculation. + // Without reentrancy detection this recurses until the stack overflows. + reactive { + val v = s() + s.value = v + 1 + } + val captured = expectException() + assertTrue( + captured is ReactiveReentrancyException, + "Expected a ReactiveReentrancyException, but captured: $captured" + ) + + } + } + + @Test + fun partialSettlingReentrancy() { + testContext { + val s = Signal(0) + reactive(reentrancyLimit = 2) { + val v = s() + if(v > 5) return@reactive + s.value = v + 1 + } + val captured = expectException() + assertTrue( + captured is ReactiveReentrancyException, + "Expected a ReactiveReentrancyException, but captured: $captured" + ) + } + } + + @Test + fun settlingReentrancy() { + testContext { + val s = Signal(0) + reactive(reentrancyLimit = 10) { + val v = s() + if(v > 5) return@reactive + s.value = v + 1 + } + assertEquals(ReactiveState(6), s.state) + } + } + + @Test + fun partialSettlingCoReentrancy() { + testContext { + val a = Signal(0) + val b = Signal(0) + reactive(reentrancyLimit = 3) { + val other = b().also { println("b is $it") } + if (other < 10) + a.value = other + 1 + } + reactive(reentrancyLimit = 3) { + val other = a().also { println("a is $it") } + if (other < 10) + b.value = other + 1 + } + val captured = expectException() + assertTrue( + captured is ReactiveReentrancyException, + "Expected a ReactiveReentrancyException, but captured: $captured" + ) + } + } + + @Test + fun settlingCoReentrancy() { + testContext { + val a = Signal(0) + val b = Signal(0) + reactive(reentrancyLimit = 10) { + val other = b().also { println("b is $it") } + if (other < 10) + a.value = other + 1 + } + reactive(reentrancyLimit = 10) { + val other = a().also { println("a is $it") } + if (other < 10) + b.value = other + 1 } - } finally { - Reactive.reportException = previous } - assertTrue( - captured.any { it is ReactiveReentrancyException }, - "Expected a ReactiveReentrancyException, but captured: $captured" - ) } @Test From fe26105cf193b8d21ff97d37ad70d2f78f289a50 Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Wed, 29 Jul 2026 11:32:38 -0600 Subject: [PATCH 11/12] More polish --- .../reactive/context/ReactiveContext.kt | 117 +++++++-- .../reactive/core/MutableRemember.kt | 2 +- .../lightningkite/reactive/core/Remember.kt | 16 +- .../com/lightningkite/reactive/deprecated.kt | 2 +- .../lightningkite/reactive/ReactivityTests.kt | 227 ++++++++++++++++-- 5 files changed, 310 insertions(+), 54 deletions(-) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt index 57e4032..34a8eae 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/context/ReactiveContext.kt @@ -132,9 +132,25 @@ typealias ReactiveContext = TypedReactiveContext<*> * context.startCalculation() * ``` * + * ### Self-Triggering Calculations + * + * A calculation that changes one of its own dependencies - most often by writing to a signal it + * also reads - asks to be rerun while it is still running. [reentrancyLimit] decides what happens: + * + * - `0` (the default) treats it as the mistake it usually is, reporting a + * [ReactiveReentrancyException] as this context's state. + * - A positive limit lets the calculation *settle*: it reruns in place, publishing each + * intermediate result, until it stops changing its own inputs or the limit is spent. Use this + * only for calculations that visibly converge, such as one that clamps a value into range. + * + * Either way the calculation never recurses, so a runaway cycle surfaces as a named error rather + * than a stack overflow (or, under a dispatching scheduler, a silent livelock). + * * @param T The type of value produced by the calculation. * @property scope The coroutine scope for calculations. * @property useLastWhileLoading Whether to preserve the last value during recalculation (true) or show loading state (false). + * @property reentrancyLimit How many times the calculation may trigger its own re-execution before + * it is reported as a [ReactiveReentrancyException]. Zero, the default, forbids self-triggering. * @property reportTo The underlying [RawReactive] to report state updates to. * @property action The calculation logic to execute in this context. */ @@ -147,9 +163,22 @@ class TypedReactiveContext( ) : DependencyChangeListener(), ReactiveCoroutineScope, Reactive by reportTo { companion object + /** + * Whether this context is currently active and tracking dependencies. + * Set to false when [cancel] is called. + */ + var active = false + private set + /** * Reference to [startCalculation] used as a listener callback. * Dependencies invoke this when they change to trigger recalculation. + * + * Invariant: a dependency registered *during* a run must not use this to deliver its initial + * value. It must hand that value to the run that registered it instead - see how the `async`, + * `Deferred.invoke` and `Flow.invoke` operators below arrange to do so. A value that first + * arrives through this path while the calculation is still running is indistinguishable from + * the calculation triggering itself, and so spends [reentrancyLimit]. */ val rerun: () -> Unit = ::startCalculation @@ -171,39 +200,56 @@ class TypedReactiveContext( */ override val coroutineContext: CoroutineContext get() = scope.coroutineContext + job + this + /** + * A recalculation has been requested but not yet performed. Set when a dependency changes and + * cleared as each run begins, so that several changes arriving before the run happens collapse + * into one, and so that a change arriving *during* a run is picked up by the settling loop in + * [startCalculation] rather than recursing. + */ private var queued = false - private var desired = false + + /** + * True while this context's [action] is executing. A recalculation requested while this is set + * came from the calculation itself - directly, by writing a signal it also reads, or by way of + * something it triggered. That is what [reentrancyLimit] governs. + */ private var calculating = false /** * Starts or restarts the reactive calculation. * - * This method: - * 1. Cancels the previous calculation's job (if any) - * 2. Creates a fresh job for this calculation run - * 3. Executes the calculation on the scope's thread - * 4. Tracks dependencies accessed during execution - * 5. Updates the reactive state with the result - * 6. Cleans up unused dependencies - * - * Thread safety: Uses [queued] flag to prevent multiple simultaneous executions. + * When no calculation is running, this cancels the previous run's coroutines, then runs the + * calculation on the scope's thread: dependencies accessed during [action] are tracked, the + * result is reported as this context's state, and dependencies that went unused are released. + * + * When a calculation *is* running, the request is left for that run's settling loop instead of + * recursing into a new one. The calculation reruns in place until it stops re-triggering + * itself, publishing each intermediate result as it goes. [reentrancyLimit] bounds how many + * times it may do so; beyond that the context reports a [ReactiveReentrancyException] rather + * than recursing until the stack overflows, or livelocking under a dispatching scheduler. The + * default limit of zero forbids self-triggering outright. */ fun startCalculation() { - desired = true - if (queued) return // Prevent duplicate queuing + active = true + if (queued) return // Prevent duplicate queuing; the pending run will see the new state queued = true + // Requested from inside the calculation. The settling loop below picks this up on its next + // pass; we must not cancel `job` here, as it holds the coroutines of the run in progress. if (calculating) return + // The previous run's inputs have changed, so anything it launched is now producing stale + // results. Cancel at the moment of the change rather than when the rerun is dispatched. + endRun() + scope.onThread { - if (!desired) return@onThread // Check if cancelled while queued + if (!active) return@onThread // Check if cancelled while queued - var iter = 0 - while(queued) { + var runs = 0 + while (queued) { queued = false - job.cancel() - job = Job() - if (iter++ > reentrancyLimit) { - reportTo.state = ReactiveState.exception(ReactiveReentrancyException(this)) + if (runs++ > reentrancyLimit) { + reportTo.state = + ReactiveState.exception(ReactiveReentrancyException(this, reentrancyLimit)) break } try { @@ -217,10 +263,19 @@ class TypedReactiveContext( dependencyBlockEnd() // Clean up dependencies not used in this run calculating = false } + // The calculation re-triggered itself, so this run's coroutines are stale for the + // same reason any other superseded run's are. + if (queued) endRun() } } } + /** Cancels the current run's coroutines and opens a fresh job for the next run. */ + private fun endRun() { + job.cancel() + job = Job() + } + init { // Automatically cancel when parent scope is cancelled scope.onRemove { cancel() } @@ -254,9 +309,8 @@ class TypedReactiveContext( * After cancellation, the context will not respond to dependency changes. */ override fun cancel() { - job.cancel() - job = Job() - desired = false + endRun() + active = false queued = false super.cancel() // Cancel dependency listeners } @@ -710,6 +764,10 @@ class TypedReactiveContext( * When the outer signal changes, both the outer and inner contexts are cancelled and recreated. * When only the inner signal changes, only the inner context reruns. * + * @param reentrancyLimit How many times [action] may trigger its own re-execution - by changing a + * dependency it also reads - before the calculation is reported as failed with a + * [ReactiveReentrancyException]. Leave at zero unless the calculation is meant to settle over + * several runs; see [TypedReactiveContext] for the trade-off. * @param action The calculation logic to run reactively * @return A [TypedReactiveContext] managing the calculation and its dependencies * @@ -791,8 +849,15 @@ object ReactiveLoading : Throwable() * signal it also reads inside the same [reactive] block. This would otherwise recurse until the * stack overflows (or livelock under a dispatching scheduler), so it is surfaced as a clear error. */ -class ReactiveReentrancyException(context: ReactiveContext) : IllegalStateException( - "A reactive calculation triggered its own re-execution ($context). This usually means the " + - "calculation wrote to a signal it also reads. Break the cycle so the calculation does " + - "not mutate its own dependencies." +class ReactiveReentrancyException(context: ReactiveContext, reentrancyLimit: Int) : IllegalStateException( + if (reentrancyLimit <= 0) + "A reactive calculation triggered its own re-execution ($context). This usually means the " + + "calculation wrote to a signal it also reads. Break the cycle so the calculation " + + "does not mutate its own dependencies. If it is instead meant to settle over " + + "several runs, raise reentrancyLimit to the number of reruns it needs." + else + "A reactive calculation kept triggering its own re-execution and did not settle within " + + "its reentrancyLimit of $reentrancyLimit reruns ($context). Either it never " + + "converges - every run changes a signal it reads, so there is no stable value to " + + "reach - or it genuinely needs a higher limit." ) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt index 7d4986b..f570fb6 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/MutableRemember.kt @@ -67,7 +67,7 @@ class MutableRemember( var overridden: Boolean = false private set - private val remember = Remember(coroutineContext, useLastWhileLoading, deactivationDelay, initialValue) + private val remember = Remember(coroutineContext, useLastWhileLoading, deactivationDelay, action = initialValue) private var forget: (()->Unit)? = null private fun startListening() { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt index 1a0aa1e..878bc93 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt @@ -20,6 +20,8 @@ import kotlin.time.Duration * @param coroutineContext The coroutine context for running the calculation (default: Dispatchers.Unconfined). * @param useLastWhileLoading If true, uses the last known value while recalculating. * @param deactivationDelay If provided, the reactive context will be kept alive for that duration after all listeners have unsubscribed. + * @param reentrancyLimit How many times [action] may trigger its own re-execution before the value + * is reported as failed; see [TypedReactiveContext]. * @param action The block to compute the value reactively. * @return A [Reactive] value that updates automatically. * @@ -41,9 +43,10 @@ fun remember( coroutineContext: CoroutineContext = Dispatchers.Unconfined, useLastWhileLoading: Boolean = false, deactivationDelay: Duration? = null, + reentrancyLimit: Int = 0, action: ReactiveContext.() -> T, ): Reactive = - Remember(coroutineContext, useLastWhileLoading, deactivationDelay, action) + Remember(coroutineContext, useLastWhileLoading, deactivationDelay, reentrancyLimit, action) /** * A reactive value that remembers the result of a calculation and shares the result among its listeners. @@ -61,6 +64,8 @@ fun remember( * @param coroutineContext The coroutine context in which the calculation runs. Defaults to [Dispatchers.Unconfined]. * @param useLastWhileLoading If true, the last known value will be used while the calculation is loading or re-running. * @param deactivationDelay If provided, the reactive context will be kept alive for that duration after all listeners have unsubscribed. + * @param reentrancyLimit How many times [action] may trigger its own re-execution before the value + * is reported as failed; see [TypedReactiveContext]. * @param action The block of code to execute within the [ReactiveContext] to produce the value. * * This class manages its own coroutine job and calculation scope. When activated, it starts the calculation @@ -73,6 +78,7 @@ class Remember( val incomingCoroutineContext: CoroutineContext = Dispatchers.Unconfined, private val useLastWhileLoading: Boolean = false, private val deactivationDelay: Duration? = null, + private val reentrancyLimit: Int = 0, private val action: ReactiveContext.() -> T, ) : Reactive, CoroutineScope, BaseListenable() { @@ -95,7 +101,13 @@ class Remember( // Starts notActive rather than notReady: nothing is listening yet, so there is no value to // be had, as opposed to one that is on its way. private val reported = RawReactive(ReactiveState.notActive) - private val scope = TypedReactiveContext(this, useLastWhileLoading, 0, reported, action) + private val scope = TypedReactiveContext( + scope = this, + useLastWhileLoading = useLastWhileLoading, + reentrancyLimit = reentrancyLimit, + reportTo = reported, + action = action, + ) // A Remember only calculates while it has listeners, and reports notActive when it has none. // It deliberately does not calculate on demand: doing so would either subscribe to sources diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/deprecated.kt b/src/commonMain/kotlin/com/lightningkite/reactive/deprecated.kt index a2cdfb9..9edca9a 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/deprecated.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/deprecated.kt @@ -99,7 +99,7 @@ typealias LateInitProperty = LateInitSignal @Deprecated("Use remember", ReplaceWith("remember", "com.lightningkite.reactive.core")) fun shared(coroutineContext: CoroutineContext = Dispatchers.Unconfined, useLastWhileLoading: Boolean = false, action: ReactiveContext.() -> T): Reactive = - remember(coroutineContext, useLastWhileLoading, null, action) + remember(coroutineContext, useLastWhileLoading, null, action = action) @Deprecated("Use reactiveProcess", ReplaceWith("reactiveProcess", "com.lightningkite.reactive.core")) fun sharedProcess(scope: CoroutineScope = AppScope, emitter: suspend Emitter.() -> Unit): Reactive = reactiveProcess(scope, emitter) diff --git a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt index 34ae9e6..e27ca6f 100644 --- a/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt +++ b/src/commonTest/kotlin/com/lightningkite/reactive/ReactivityTests.kt @@ -26,6 +26,7 @@ import com.lightningkite.reactive.core.Signal import com.lightningkite.reactive.core.remember import kotlinx.coroutines.* import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.test.advanceTimeBy import kotlinx.coroutines.test.runCurrent import kotlinx.coroutines.test.runTest @@ -517,8 +518,24 @@ class ReactivityTests { captured is ReactiveReentrancyException, "Expected a ReactiveReentrancyException, but captured: $captured" ) + } + } + @Test + fun reentrancyErrorSaysWhichProblemItIs() { + // A calculation that opted into settling needs to hear that its budget ran out. Telling it + // to stop writing its own dependencies is advice it has already declined. + val unwanted = testContext { + val s = Signal(0) + reactive { val v = s(); s.value = v + 1 } + expectException().message } + val unsettled = testContext { + val s = Signal(0) + reactive(reentrancyLimit = 3) { val v = s(); s.value = v + 1 } + expectException().message + } + assertNotEquals(unwanted, unsettled) } @Test @@ -527,14 +544,10 @@ class ReactivityTests { val s = Signal(0) reactive(reentrancyLimit = 2) { val v = s() - if(v > 5) return@reactive + if (v > 5) return@reactive s.value = v + 1 } - val captured = expectException() - assertTrue( - captured is ReactiveReentrancyException, - "Expected a ReactiveReentrancyException, but captured: $captured" - ) + assertIs(expectException()) } } @@ -542,12 +555,149 @@ class ReactivityTests { fun settlingReentrancy() { testContext { val s = Signal(0) + val ctx = reactive(reentrancyLimit = 10) { + val v = s() + if (v > 5) return@reactive + s.value = v + 1 + } + assertEquals(ReactiveState(6), s.state) + assertNull(ctx.state.exception, "the calculation settled, so it should not have failed") + } + } + + @Test + fun settlingExactlyAtTheLimitSucceeds() { + // Climbing to six takes six writes, so six self-triggered reruns: precisely the budget. + testContext { + val s = Signal(0) + val ctx = reactive(reentrancyLimit = 6) { + val v = s() + if (v > 5) return@reactive + s.value = v + 1 + } + assertEquals(ReactiveState(6), s.state) + assertNull(ctx.state.exception) + } + } + + @Test + fun settlingOneShortOfTheLimitFails() { + testContext { + val s = Signal(0) + reactive(reentrancyLimit = 5) { + val v = s() + if (v > 5) return@reactive + s.value = v + 1 + } + assertIs(expectException()) + } + } + + @Test + fun settlingPublishesEachIntermediateValue() { + // Deliberate: a settling calculation publishes every step rather than only its fixed point. + // That is what lets two settling calculations drive each other along, as the co-reentrancy + // tests below rely on. + testContext { + val s = Signal(0) + val observed = ArrayList() + reactive { observed.add(s()) } reactive(reentrancyLimit = 10) { val v = s() - if(v > 5) return@reactive + if (v > 5) return@reactive s.value = v + 1 } + assertEquals(listOf(0, 1, 2, 3, 4, 5, 6), observed) + } + } + + @Test + fun eachExternalChangeGetsAFreshReentrancyBudget() { + // The limit bounds one settling sequence, not the lifetime of the calculation. A + // calculation that self-triggers once per change has to keep working indefinitely. + testContext { + val trigger = Signal(0) + val clamped = Signal(0) + val ctx = reactive(reentrancyLimit = 1) { + val t = trigger() + if (clamped() != t) clamped.value = t + } + repeat(20) { trigger.value = it + 1 } + assertEquals(ReactiveState(20), clamped.state) + assertNull(ctx.state.exception) + } + } + + @Test + fun aContextRecoversAfterAReentrancyError() { + // The failed run keeps the dependencies of its last complete pass, so a later change still + // reaches the calculation and it computes normally once the cycle is gone. + testContext { + val cycle = Signal(0) + val other = Signal("a") + var runaway = true + val ctx = reactive { + val o = other() + if (runaway) cycle.value = cycle() + 1 + o + } + assertIs(expectException()) + + runaway = false + other.value = "b" + assertEquals(ReactiveState("b"), ctx.state) + } + } + + @Test + fun aFlowsFirstValueIsNotReentrancy() { + // A cold flow collects synchronously under an undispatched scheduler, so its first value + // lands while the calculation that registered it is still running. That is a dependency + // arriving, not the calculation triggering itself, so it must not spend the budget. + testContext { + val flow = flowOf(7) + val seen = ArrayList() + val ctx = reactive { seen.add(flow()) } + assertEquals(listOf(7), seen) + assertNull(ctx.state.exception) + } + } + + @Test + fun aStateFlowsCurrentValueIsNotReentrancy() { + testContext { + val flow = MutableStateFlow(1) + val seen = ArrayList() + val ctx = reactive { seen.add(flow()) } + flow.value = 2 + assertEquals(listOf(1, 2), seen) + assertNull(ctx.state.exception) + } + } + + @Test + fun rememberDefaultsToRejectingReentrancy() { + testContext { + val s = Signal(0) + val r = remember { val v = s(); s.value = v + 1; v } + reactive { r() } + assertIs(expectException()) + } + } + + @Test + fun rememberPassesItsReentrancyLimitThrough() { + testContext { + val s = Signal(0) + val settled = remember(reentrancyLimit = 10) { + val v = s() + if (v <= 5) s.value = v + 1 + v + } + val seen = ArrayList() + reactive { seen.add(settled()) } assertEquals(ReactiveState(6), s.state) + assertEquals(6, seen.last()) } } @@ -557,38 +707,42 @@ class ReactivityTests { val a = Signal(0) val b = Signal(0) reactive(reentrancyLimit = 3) { - val other = b().also { println("b is $it") } - if (other < 10) - a.value = other + 1 + val other = b() + if (other < 10) a.value = other + 1 } reactive(reentrancyLimit = 3) { - val other = a().also { println("a is $it") } - if (other < 10) - b.value = other + 1 + val other = a() + if (other < 10) b.value = other + 1 } - val captured = expectException() - assertTrue( - captured is ReactiveReentrancyException, - "Expected a ReactiveReentrancyException, but captured: $captured" - ) + assertIs(expectException()) } } @Test fun settlingCoReentrancy() { + // Two calculations driving each other converge as long as each one's own budget covers the + // reruns it personally performs. They are not symmetric: the first context is re-entered + // from scratch by the second - so its budget is never spent - while the second settles in + // place and pays for every step. testContext { val a = Signal(0) val b = Signal(0) + val firstSaw = ArrayList() + val secondSaw = ArrayList() reactive(reentrancyLimit = 10) { - val other = b().also { println("b is $it") } - if (other < 10) - a.value = other + 1 + val other = b() + firstSaw.add(other) + if (other < 10) a.value = other + 1 } reactive(reentrancyLimit = 10) { - val other = a().also { println("a is $it") } - if (other < 10) - b.value = other + 1 + val other = a() + secondSaw.add(other) + if (other < 10) b.value = other + 1 } + assertEquals(ReactiveState(9), a.state) + assertEquals(ReactiveState(10), b.state) + assertEquals(listOf(0, 2, 4, 6, 8, 10), firstSaw) + assertEquals(listOf(1, 3, 5, 7, 9), secondSaw) } } @@ -664,6 +818,31 @@ class ReactivityTests { ctx.cancel() } + @Test + fun changingADependencyEndsThePreviousRunImmediately() = runTest { + // Under a dispatching scheduler the rerun does not begin until the scheduler gets to it, + // but the run it supersedes is finished the moment its input changed - whatever that run + // launched is now computing from a stale value. Cleanup must not wait for the rerun. + val trigger = Signal(0) + val cleanedUp = ArrayList() + val ctx = TypedReactiveContext(this) { + val t = trigger() + onRemove { cleanedUp.add(t) } + } + ctx.startCalculation() + runCurrent() + assertEquals(listOf(), cleanedUp, "the first run is still current") + + trigger.value = 1 + assertEquals( + listOf(0), cleanedUp, + "the superseded run should be torn down at the change, not when the rerun is dispatched" + ) + + runCurrent() + ctx.cancel() + } + @Test fun readingStateWithoutListenersDoesNotCalculate() { val source = Signal(1) From cd0eb8f8e18829da0f550866a78fc35810b5e9a3 Mon Sep 17 00:00:00 2001 From: UnknownJoe796 Date: Wed, 29 Jul 2026 13:22:39 -0600 Subject: [PATCH 12/12] Changes made as requested by Hunter Backported from reactive-api-split so the two branches agree on behavior: - ReactiveState.ready compares the sentinels with != rather than !is. The sentinels are objects, so equality is the intended check. - Remember.activate no longer overwrites notActive with notReady when useLastWhileLoading is set - that is the whole point of the flag. - onNextSuccess routes both the listener and the post-subscribe read through one `perform` guarded by `acted`, instead of duplicating the body and reconciling afterwards. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01GA6LAUN9Lpo5hyK3KVJteJ --- .../reactive/core/ReactiveState.kt | 2 +- .../lightningkite/reactive/core/Remember.kt | 2 +- .../reactive/extensions/helpers.kt | 21 +++++++------------ 3 files changed, 9 insertions(+), 16 deletions(-) diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt index 88d9986..da8ad4a 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/ReactiveState.kt @@ -29,7 +29,7 @@ import kotlin.jvm.JvmInline @JvmInline @OptIn(InternalReactiveApi::class) value class ReactiveState(val raw: T) { - inline val ready: Boolean get() = raw !is InternalReactiveNotReady && raw !is InternalReactiveNotActive + inline val ready: Boolean get() = raw != InternalReactiveNotReady && raw != InternalReactiveNotActive inline val success: Boolean get() = ready && raw !is InternalReactiveThrownException /** True when nothing is maintaining this value; see the [ReactiveState] docs. */ diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt b/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt index 878bc93..d0134a5 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/core/Remember.kt @@ -130,7 +130,7 @@ class Remember( // Something is maintaining this value again - it just doesn't have one yet. Without this, // useLastWhileLoading would suppress the notReady the first calculation reports and leave // notActive in place, claiming nobody is listening when somebody now is. - if (reported.state.notActive) reported.state = ReactiveState.notReady + if (reported.state.notActive && !useLastWhileLoading) reported.state = ReactiveState.notReady shuttingDown?.let { CoroutineScope(incomingCoroutineContext).launch { diff --git a/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt b/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt index 3faca02..1aae25a 100644 --- a/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt +++ b/src/commonMain/kotlin/com/lightningkite/reactive/extensions/helpers.kt @@ -69,23 +69,16 @@ fun Reactive.onNextSuccess(action: (T) -> Unit): Release? { var release: Release? = null var acted = false - release = addListener { - state.onSuccess { - acted = true - action(it) - release?.invoke() - } - } - // Read after subscribing: activating a notActive source calculates inside addListener, before - // our listener is in place. - state.onSuccess { + fun perform(value: T) { + if (acted) return + release?.invoke() acted = true - action(it) + action(value) } - if (acted) { - release?.invoke() - return null + release = addListener { + state.onSuccess(::perform) } + state.onSuccess(::perform) return release }