Merge pull request #821 from gemini-hlsw/revision

Add tests, improve execution of throtting logic, fix README, fix stale effects in useEffectResult

Author: rpiaggio

CLAUDE.md

@@ -0,0 +1,64 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## Overview
+
+`crystal` is a Scala library that provides a toolbelt for building reactive UI apps in Scala. It
+targets Scala 3 and cross-compiles to both the JVM and Scala.js. Publishing (organization,
+versioning, Maven Central release) is handled by the `sbt-lucuma-lib` plugin under the `edu.gemini`
+organization — the `com.rpiaggio` coordinate in the README badge is outdated. The README has
+extensive, authoritative API documentation — read it before changing public APIs.
+
+Core abstractions:
+- **`Pot[A]` / `PotOption[A]`** (`modules/core/shared`): sum types for delayed/requested values
+  (`Pending` / `Ready` / `Error`, plus `ReadyNone` for `PotOption`).
+- **`ViewF[F, A]`** and variants `ViewOptF` / `ViewListF` (`viewF.scala`): a value plus an
+  effectful modify callback `(A => A) => F[Unit]`, with `zoom` (via Monocle optics) to drill into
+  substructure. `View[A]` is the `Callback`-fixed alias used by the react layer.
+- **`Reuse[A]`** (`modules/core/js`): attaches a hidden `B` with a `Reusability[B]` to a value so
+  that types without a natural `Reusability` (functions, `VdomNode`) can participate in reuse.
+- **Throttling**: `Throttler`, `ThrottlingViewF`, `ViewThrottlerF`, `Deglitcher` coordinate UI vs.
+  server-driven updates.
+- **scalajs-react hooks** (`modules/core/js/.../react/hooks`): monadic hooks integrating
+  `cats-effect` `IO` and `fs2.Stream` with scalajs-react (e.g. `useSingleEffect`, `useStreamView`,
+  `useStreamResource`, `useEffectStream`, `useSignalStream`). New hooks are added only in their
+  monadic form; builder-style is being phased out.
+
+The library assumes scalajs-react's `core-bundle-cb_io` bundle (sync effect `CallbackTo`, async
+effect `IO`).
+
+## Module structure
+
+Three cross-projects (JVM + JS), defined in `build.sbt`:
+- **`core`** (`modules/core`): the library. `shared/` holds platform-agnostic code (`Pot`, `ViewF`,
+  throttling); `js/` holds the scalajs-react integration (`react/`, hooks, `Reuse`). There is no
+  meaningful `jvm/`-only source — the JS layer is where react lives.
+- **`testkit`** (`modules/testkit`): ScalaCheck `Arbitrary`/`Eq` instances for the core types,
+  published for downstream test use. Depends on `core`.
+- **`tests`** (`modules/tests`): the test suite (MUnit + discipline law checks). Not published.
+  JS tests run in a JSDOM environment. Depends on `testkit`.
+
+## Build, test, lint
+
+The build uses sbt via the bundled `sbt-launch.jar` and the `sbt-lucuma-lib` plugin (which derives
+versioning, CI config, scalafmt/scalafix config, and publishing). Dependency versions are
+centralized in `project/Settings.scala`. JS tests require `npm ci` first (provides `jsdom`, `react`,
+`react-dom`).
+
+```bash
+sbt compile                  # compile all (JVM + JS)
+sbt test                     # run all tests (JVM + JS)
+sbt rootJVM/test             # only JVM tests
+sbt rootJS/test              # only JS tests (requires `npm ci` first)
+sbt "tests / Test / testOnly crystal.PotSpec"   # a single test suite
+sbt scalafmtAll              # format
+sbt scalafmtCheckAll         # verify formatting (CI gate)
+sbt "scalafixAll OrganizeImports"   # organize imports (the configured scalafix rule)
+```
+
+Notes:
+- Scala 3 only (`3.3.8` cross-version in `build.sbt`). `maxColumn = 100`; vertical alignment is on.
+- Don't hand-edit `.scalafmt*.conf` / `.scalafix*.conf` — they are generated by `sbt-lucuma`.
+- CI (`.github/workflows/ci.yml`) sets up Node 24 + `npm ci` only for the `rootJS` matrix project.
+- Both `core` and `testkit` are published; treat their public APIs as compatibility-sensitive.

README.md

@@ -24,7 +24,7 @@
           imports = [ typelevel-nix.typelevelShell ];
           typelevelShell = {
             nodejs.enable = true;
-            jdk.package = pkgs.jdk17;
+            jdk.package = pkgs.jdk25;
           };
         };
       }

flake.nix

@@ -3,6 +3,7 @@
 
 package crystal.react.hooks
 
+import cats.effect.Outcome
 import cats.syntax.all.*
 import crystal.*
 import crystal.react.*
@@ -31,30 +32,42 @@ object UseEffectResult:
     val depsOpt: Option[D] = effect.deps.toOption
 
   private def doRefresh[A](
-    effect:     DefaultA[A],
-    set:        Pot[A] => DefaultS[Unit],
-    setRunning: Boolean => DefaultS[Unit]
+    effect:       DefaultA[A],
+    set:          Pot[A] => DefaultS[Unit],
+    setRunning:   Boolean => DefaultS[Unit],
+    singleEffect: UseSingleEffect[DefaultA]
   )(
-    keep:       Boolean
+    keep:         Boolean
   ): DefaultS[Unit] =
     set(Pot.pending).unless(keep).void >>
       setRunning(true) >>
-      (effect >>= (a => set(a.ready).toAsync))
-        .handleErrorWith(t => set(Pot.error(t)).toAsync)
-        .guarantee(setRunning(false).toAsync)
+      // Route through `singleEffect` so a new refresh (deps change or manual) cancels the previous
+      // in-flight effect, preventing a stale result from overwriting a newer one. `setRunning(false)`
+      // must not run on cancellation: the new run already set `isRunning` to `true`, so a cancelled
+      // run clearing it would leave the flag wrong for the duration of the new effect.
+      singleEffect
+        .submit:
+          (effect >>= (a => set(a.ready).toAsync))
+            .handleErrorWith(t => set(Pot.error(t)).toAsync)
+            .guaranteeCase:
+              case Outcome.Canceled() => ().pure[DefaultA]
+              case _                  => setRunning(false).toAsync
         .runAsyncAndForget
 
   // Provides functionality for all the flavors
   private def hookBuilder[D, A, R: Reusability](
     deps: Pot[D]
   )(effect: D => DefaultA[A], keep: Boolean, reuseBy: Option[R]): HookResult[UseEffectResult[A]] =
     for
-      state     <- useSerialStateView(Pot.pending[A])
-      isRunning <- useState(false)
-      effectOpt <- useMemo(reuseBy): _ => // Memo Option[effect]
-                     deps.toOption.map(effect)
-      refresh    = effectOpt.map(_.foldMap(doRefresh(_, state.set, isRunning.setState)))
-      _         <- useEffectWithDeps(refresh)(_(keep))
+      state        <- useSerialStateView(Pot.pending[A])
+      isRunning    <- useState(false)
+      singleEffect <- useSingleEffect
+      effectOpt    <- useMemo(reuseBy): _ => // Memo Option[effect]
+                        deps.toOption.map(effect)
+      refresh       = effectOpt.map(
+                        _.foldMap(doRefresh(_, state.set, isRunning.setState, singleEffect.value))
+                      )
+      _            <- useEffectWithDeps(refresh)(_(keep))
     yield UseEffectResult(state.map(_.toPotView), isRunning.value, refresh, keep)
 
   /**

modules/core/js/src/main/scala/crystal/react/hooks/UseEffectResult.scala

@@ -17,26 +17,38 @@ import scala.concurrent.duration.FiniteDuration
  * be discarded if they are submitted while another effect is running, except for the last one.
  */
 class Throttler[F[_]: Temporal] private (
-  spacedBy:  FiniteDuration,
-  isRunning: Ref[F, Boolean], // true if an effect is running
-  queued:    Ref[F, Option[F[Unit]]]
+  spacedBy: FiniteDuration,
+  state:    Ref[F, Throttler.State[F]] // running flag and last queued effect, updated atomically
 )(using Monoid[F[Unit]]):
+  import Throttler.State
+
   def submit(f: F[Unit]): F[Unit] =
-    (queued.set(none) >> isRunning.getAndSet(true)).uncancelable
-      .flatMap:
-        case false =>
-          (Temporal[F].sleep(spacedBy).both(f) >> // Completes when both complete.
-            (isRunning.set(false) >> queued.getAndSet(none)).uncancelable
-              .flatMap(_.map(submit).orEmpty)).start.void // Execute everything in background.
-        case true  =>
-          queued.set(f.some)
+    state
+      .modify: // If not running, start now; otherwise queue f.
+        case State(false, _) => (State(true, none), startRun(f))
+        case State(true, _)  => (State(true, f.some), Monoid[F[Unit]].empty)
+      .flatten
+
+  // Runs `f`, spaced by at least `spacedBy`, in the background; then drains the last queued effect.
+  private def startRun(f: F[Unit]): F[Unit] =
+    (Temporal[F].sleep(spacedBy).both(f) >> // Completes when both complete.
+      drain).start.void                     // Execute everything in background.
+
+  // Atomically clears the running flag, re-submitting the last queued effect, if any.
+  private val drain: F[Unit] =
+    state
+      .modify: // If there is a queued effect, start it; otherwise, clear the running flag.
+        case State(_, Some(next)) => (State(true, none), startRun(next))
+        case State(_, None)       => (State(false, none), Monoid[F[Unit]].empty)
+      .flatten
 
 object Throttler:
+  // `isRunning` and `queuedEffect` are kept in a single `Ref` so that "is something running?" and "enqueue
+  // the last effect" happen as one atomic transition.
+  private case class State[F[_]](isRunning: Boolean = false, queuedEffect: Option[F[Unit]] = none)
+
   def apply[F[_]: Temporal](spacedBy: FiniteDuration)(using Monoid[F[Unit]]): F[Throttler[F]] =
-    for
-      isRunning <- Ref.of(false)
-      queued    <- Ref.of(none)
-    yield new Throttler(spacedBy, isRunning, queued)
+    Ref.of[F, State[F]](State()).map(new Throttler(spacedBy, _))
 
   def unsafe[F[_]: Temporal: Sync](spacedBy: FiniteDuration)(using Monoid[F[Unit]]): Throttler[F] =
-    new Throttler(spacedBy, Ref.unsafe(false), Ref.unsafe(none))
+    new Throttler(spacedBy, Ref.unsafe(State()))

modules/core/shared/src/main/scala/crystal/Throttler.scala

@@ -8,6 +8,8 @@ import cats.effect.Temporal
 import cats.effect.syntax.all.*
 import cats.syntax.all.*
 import cats.~>
+import monocle.Focus
+import monocle.Lens
 
 import scala.concurrent.duration.FiniteDuration
 
@@ -28,39 +30,40 @@ import scala.concurrent.duration.FiniteDuration
  *   G The async effect. This will be used for concurrency.
  */
 final class ViewThrottlerF[F[_], G[_], A] private (
-  waitUntil:     Ref[G, FiniteDuration],
-  nextUpdate:    Ref[G, Option[ViewThrottlerF.ModCBType[G, A]]],
+  state:         Ref[G, ViewThrottlerF.State[G, A]],
   timeout:       FiniteDuration,
   syncToAsync:   F ~> G,
   dispatchAsync: G[Unit] => F[Unit]
 )(using G: Temporal[G]) {
+  import ViewThrottlerF.State
 
   private def throttle: F[Unit] =
     dispatchAsync:
-      G.monotonic.flatMap(now => waitUntil.set(now + timeout))
+      G.monotonic.flatMap(now => state.update(State.waitUntil[G, A].replace(now + timeout)))
 
   private def throttlerView(view: ViewF[F, A]): ViewF[F, A] =
     view.withOnMod(_ => throttle)
 
   private def attemptSet(modCB: (A => A, (A, A) => G[Unit]) => G[Unit]): G[Unit] =
-    (waitUntil.get, G.monotonic).flatMapN: (waitUntil, now) =>
-      if (waitUntil > now)
-        (G.sleep(waitUntil - now) >> attemptSet(modCB)).start.void
-      else
-        nextUpdate
-          .flatModify: accumModCb =>
-            (none, accumModCb.map((mod, cb) => modCB(mod, cb)).getOrElse(G.unit))
-          .flatTap(_ => G.unit)
+    G.monotonic.flatMap: now =>
+      state
+        .modify: s =>
+          if (s.waitUntil > now)
+            (s, (G.sleep(s.waitUntil - now) >> attemptSet(modCB)).start.void)
+          else
+            (State.nextUpdate[G, A].replace(none)(s),
+             s.nextUpdate.map((mod, cb) => modCB(mod, cb)).getOrElse(G.unit)
+            )
+        .flatten
 
   private def throttledView(view: ViewF[F, A]): ViewF[G, A] =
     new ViewF[G, A](
       get = view.get,
       modCB = (mod, cb) =>
-        nextUpdate.update: x =>
-          x match
-            case None            => (mod, cb).some
-            case Some(oldMod, _) => (oldMod.andThen(mod), cb).some // We only keep last CB
-        >>
+        state.update(State.nextUpdate[G, A].modify {
+          case None            => (mod, cb).some
+          case Some(oldMod, _) => (oldMod.andThen(mod), cb).some // We only keep last CB
+        }) >>
           attemptSet: (newMod, newCB) =>
             syncToAsync(view.modCB(newMod, (oldA, newA) => dispatchAsync(newCB(oldA, newA))))
     )
@@ -70,13 +73,25 @@ final class ViewThrottlerF[F[_], G[_], A] private (
 }
 
 object ViewThrottlerF {
+  // The type of ViewF.modCB's parameters, to avoid repeating everywhere.
   private type ModCBType[F[_], A] = (A => A, (A, A) => F[Unit])
 
+  // `waitUntil` (pause deadline) and `nextUpdate` (last accumulated update) are kept in a single
+  // `Ref` so that, in `attemptSet`, deciding whether the pause has elapsed and taking the pending
+  // update happen as one atomic transition.
+  private case class State[G[_], A](waitUntil: FiniteDuration, nextUpdate: Option[ModCBType[G, A]])
+  private object State:
+    def waitUntil[G[_], A]: Lens[State[G, A], FiniteDuration]           =
+      Focus[State[G, A]](_.waitUntil)
+    def nextUpdate[G[_], A]: Lens[State[G, A], Option[ModCBType[G, A]]] =
+      Focus[State[G, A]](_.nextUpdate)
+
   def apply[F[_], G[_], A](
     timeout:       FiniteDuration,
     syncToAsync:   F ~> G,
     dispatchAsync: G[Unit] => F[Unit]
   )(using G: Temporal[G]): G[ViewThrottlerF[F, G, A]] =
-    (G.monotonic.flatMap(G.ref(_)), G.ref(none[ModCBType[G, A]]))
-      .mapN(new ViewThrottlerF(_, _, timeout, syncToAsync, dispatchAsync))
+    G.monotonic
+      .flatMap(now => G.ref(State[G, A](now, none)))
+      .map(new ViewThrottlerF(_, timeout, syncToAsync, dispatchAsync))
 }

modules/core/shared/src/main/scala/crystal/ViewThrottlerF.scala

@@ -0,0 +1,79 @@
+// Copyright (c) 2016-2025 Association of Universities for Research in Astronomy, Inc. (AURA)
+// For license information see LICENSE or https://opensource.org/licenses/BSD-3-Clause
+
+package crystal.react.hooks
+
+import cats.effect.Deferred
+import cats.effect.IO
+import japgolly.scalajs.react.*
+import japgolly.scalajs.react.test.ReactTestUtils
+import japgolly.scalajs.react.testing_library.dom.Simulate
+import japgolly.scalajs.react.vdom.html_<^.*
+import munit.CatsEffectSuite
+import org.scalajs.dom
+
+import scala.concurrent.duration.*
+
+// Exercises `useAsyncEffect` against its README description: "Version of `useEffect` that avoids
+// race conditions when executing async effects [...] by cancelling the previous instance of the
+// effect before executing a new one."
+class UseAsyncEffectSpec extends CatsEffectSuite:
+  import ReactTestUtils.*
+
+  test("useAsyncEffect - cancels the previous effect when deps change"):
+    val log       = cats.effect.Ref.unsafe[IO, List[Int]](Nil)
+    val buttonRef = Ref[dom.HTMLButtonElement]
+
+    val comp = ScalaFnComponent[Unit]: _ =>
+      for
+        dep <- useState(0)
+        // dep 0's effect is slow; dep 1's is fast, so dep 0 would record *after* dep 1 if not
+        // cancelled on the deps change.
+        _   <- useAsyncEffectWithDeps(dep.value): d =>
+                 IO.sleep(if d == 0 then 100.millis else 10.millis) >> log.update(d :: _)
+      yield <.button("x", ^.onClick --> dep.modState(_ + 1)).withRef(buttonRef)
+
+    withRendered(comp()): _ =>
+      for
+        // No pre-click wait needed: on the deps change, `useSingleEffect` waits for dep 0's effect
+        // to have started, then cancels it mid-sleep.
+        _ <- act_(Simulate.click(buttonRef.unsafeGet())) // dep -> 1, cancels dep 0's effect
+        // Wait past dep 0's would-be completion (100ms) to confirm it never recorded.
+        _ <- act(IO.sleep(120.millis))
+        r <- log.get
+        _  = assertEquals(r, List(1))                    // only dep 1 ran; dep 0 was cancelled
+      yield ()
+
+  test("useAsyncEffect - runs the cleanup effect on deps change"):
+    val log       = cats.effect.Ref.unsafe[IO, List[String]](Nil)
+    val started1  = Deferred.unsafe[IO, Unit] // signalled when the dep-1 effect has run
+    val buttonRef = Ref[dom.HTMLButtonElement]
+
+    val comp = ScalaFnComponent[Unit]: _ =>
+      for
+        dep <- useState(0)
+        _   <- useAsyncEffectWithDeps(dep.value): d =>
+                 (log.update(s"start$d" :: _) >>
+                   (if d == 1 then started1.complete(()).void else IO.unit))
+                   .as(log.update(s"cleanup$d" :: _))
+      yield <.button("x", ^.onClick --> dep.modState(_ + 1)).withRef(buttonRef)
+
+    withRendered(comp()): _ =>
+      for
+        _ <- act_(Simulate.click(buttonRef.unsafeGet()))
+        // Wait deterministically until the dep-1 effect has run. `useSingleEffect`'s latch keeps the
+        // order, so by then `start0 -> cleanup0 -> start1` have all happened.
+        _ <- act(started1.get)
+        r <- log.get
+        _  = assertEquals(r.reverse, List("start0", "cleanup0", "start1"))
+      yield ()
+
+  test("useAsyncEffectOnMount - runs the effect on mount"):
+    val ran = Deferred.unsafe[IO, Unit]
+
+    val comp = ScalaFnComponent[Unit]: _ =>
+      for _ <- useAsyncEffectOnMount(ran.complete(()).void)
+      yield EmptyVdom
+
+    // `act(ran.get)` only completes if the on-mount effect ran.
+    withRendered(comp())(_ => act(ran.get))

modules/tests/js/src/test/scala/crystal/react/hooks/UseAsyncEffectSpec.scala

@@ -360,3 +360,29 @@ class UseEffectResultSpec extends CatsEffectSuite:
                _  = d.innerHTML.assert("(Reusable(Ready(3)),false)")
              yield ()
     yield ()
+
+  test("useEffectResultWithDeps - deps change cancels in-flight effect (no stale overwrite)"):
+    val buttonRef = Ref[dom.HTMLButtonElement]
+
+    // The first deps' effect is slow and the second deps' effect is fast, so the first one would
+    // resolve *after* the second. If the in-flight effect is not cancelled when deps change, its
+    // stale result overwrites the newer one.
+    val comp = ScalaFnComponent[Unit]: _ =>
+      for
+        s <- useState(0)
+        v <- useEffectResultWithDeps(s.value): d =>
+               if d === 0 then IO.sleep(100.millis).as(10) else IO.sleep(10.millis).as(20)
+      yield <.button((v.value, v.isRunning).toString, ^.onClick --> s.modState(_ + 1))
+        .withRef(buttonRef)
+
+    withRendered(comp()): d =>
+      d.innerHTML.assert("(Reusable(Pending),true)")
+      for
+        _ <- act(IO.sleep(20.millis))                          // slow effect (deps 0) still running
+        _ <- act_(Simulate.click(buttonRef.unsafeGet()))       // change deps -> start fast effect
+        _  = d.innerHTML.assert("(Reusable(Pending),true)")
+        _ <- act(IO.sleep(10.millis))                          // fast effect (deps 1) completes
+        _  = d.innerHTML.assert("(Reusable(Ready(20)),false)")
+        _ <- act(IO.sleep(100.millis))                         // slow effect (deps 0) would complete
+        _  = d.innerHTML.assert("(Reusable(Ready(20)),false)") // must NOT be overwritten by stale 10
+      yield ()