feature: Cross-platform Task primitive for cancellable background work

[xerial]

Summary

Adds wvlet.uni.control.Task, a cross-platform primitive for "run a unit of work on a background worker that can be cooperatively cancelled and awaited" — the requirement from #552. Lands the full design end-to-end on JVM, Scala.js (Node + browser), and Scala Native in one PR so the design itself is verified.

Sits in wvlet.uni.control alongside Retry, CircuitBreaker, RateLimiter, Guard, Resource — the existing home for lifecycle / reliability primitives.

Two API surfaces

• Task.run { body } — closure-based, works on all platforms. await() blocks on JVM/Native; throws UnsupportedOperationException on Scala.js (would deadlock the event loop) — use awaitRx.
• Task.register("id") { body } + Task.runRegistered("id") — registry-based. On Node, the body runs in a worker_threads worker so the main thread can Atomics.wait — blocking await() works on Node. Required for the wvlet cross-platform query-runner use case. On JVM/Native this is a thin alias over run; on browser it falls back to the microtask path (no main-thread Atomics.wait).

Lifecycle

• No [A] result type, no progress API. Body is TaskContext => Unit. Bodies publish anything (progress, partial results, terminal values) through any existing Rx primitive (RxVar, RxQueue, RxDeferred). Same "side-effects managed outside the interface" rule that already justified no result type.
• cancel(reason: String = "") — reason flows into the InterruptedException that checkCancelled throws and that await() / awaitRx surface.
• State machine: Running | Succeeded | Failed | Cancelled — no Cancelling (unreliable observable; a racing terminal set could win).

Node worker_threads — non-obvious bits

Captured in adr/2026-05-18-node-task-worker-threads.md:

• Bundle URL via js.\import`.meta.url, not process.argv[1]` (sbt-jsenv-nodejs leaves argv[1] empty).
• The test bundle's tail Bridge.start() throws in worker isolates because scalajsCom is undefined — the worker bootstrap tolerates the import error if globalThis.__uniTaskInvoke got set first (eagerly via an @JSExportTopLevel val).
• js.Dynamic.global.updateDynamic("foo")(v) is rewritten by the Scala.js linker as foo = v, which throws ReferenceError in strict-mode ESM. Use Object.defineProperty via js.eval("globalThis") instead.
• Bodies must be registered at module-init time so the worker's bundle re-import re-populates the registry in the worker isolate. Documented on Task.register.

Test plan

• uniJVM/testOnly *TaskFeasibilityTest *JvmTaskTest — 9/9 (lifecycle, blocking await, Thread.interrupt-driven cancel of a sleeping body, cancel(reason))
• uniJS/testOnly *TaskFeasibilityTest *JsTaskTest *NodeWorkerTaskTest — 13/13 (microtask lifecycle, browser-side await throws, 4 NodeWorkerTaskTest covering real blocking-await: success, body-exception, cancel-with-reason, missing-registration)
• uniNative/testOnly *TaskFeasibilityTest *NativeTaskTest — 10/10 (lifecycle + JVM-shape blocking/interrupt tests)
• scalafmtAll clean
• Full projectJVM/test clean (no regressions)

Design doc: plans/2026-05-18-cross-platform-thread.md. ADR: adr/2026-05-18-node-task-worker-threads.md.

🤖 Generated with Claude Code

[gemini-code-assist]

Code Review

This pull request introduces a cross-platform Task primitive for background work across JVM, Scala Native, and Scala.js, including a design document, core API traits, and platform-specific implementations. The feedback suggests enhancing type safety by parameterizing the progress type instead of using Any. For the JVM implementation, it is recommended to use a shared thread pool for efficiency. Additionally, a change to the cancel() logic is proposed to allow re-triggering the interruption signal if a task remains in the cancelling state.

[gemini-code-assist]

Using type Progress = Any forces users to use manual downcasting when consuming progress updates, which is error-prone and less idiomatic in a type-safe library like uni. Since the design doc already notes a preference for a typed approach (e.g., Task[+P]), it would be better to parameterize Task and TaskContext with a progress type P now. This would allow consumers to work with specific types (like QueryStats) directly through the progress and progressStream APIs.

[gemini-code-assist]

Creating a new Thread for every task via threadFactory.newThread is inefficient for high-frequency background work. Consider using a shared ExecutorService (e.g., Executors.newCachedThreadPool) to manage and reuse threads. This improvement is also suggested in the design doc under 'Platform implementations' for the JVM.

[gemini-code-assist]

The cancel() implementation is strictly one-shot. If a task body catches InterruptedException but continues execution (for example, if it has multiple sequential blocking calls), subsequent calls to cancel() will not re-trigger onCancelRequested(). Consider allowing onCancelRequested() to be called again if the task is still in the Cancelling state, ensuring that the interruption signal can be re-sent if the first one was insufficient.

  final override def cancel(): Unit =
    if cancelFlag.compareAndSet(false, true) then
      stateRef.compareAndSet(Task.State.Running, Task.State.Cancelling)
      onCancelRequested()
    else if state == Task.State.Cancelling then
      onCancelRequested()

[xerial]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces wvlet.uni.control.Task, a cross-platform primitive for managing background work with cooperative cancellation across JVM, Scala Native, and Scala.js. The implementation provides a unified API for lifecycle management, though the design document needs updates to reflect the removal of the progress API and generic result types. Feedback also suggests optimizing performance on JVM and Native platforms by utilizing shared thread pools instead of spawning new threads for every task.

[gemini-code-assist]

This design document appears to be partially out of sync with the final implementation. It contains several references to features that were decided against, which could be confusing for future readers.

Specifically:

• Progress API: Sections for JVM, Scala Native, and Scala.js (e.g., lines 128-129, 139, 170-172) and the 'Open questions' section (lines 238-242) still describe a built-in progress API, even though the 'Naming, shape, and package' section correctly states it was removed.
• Generic Result Type [A]: Several places (e.g., lines 207, 241, 262) mention a generic result type [A] for Task and awaitRx, but the final API uses Unit.

It would be beneficial to update the document to consistently reflect the final API surface (Task with no result type and no progress API).

[gemini-code-assist]

Creating a new thread for each Task can be inefficient, especially if many short-lived tasks are created. As noted in the design document, using a shared ExecutorService with a thread pool would be more performant by reusing threads.

Consider using a shared ExecutorService, for example, one created with java.util.concurrent.Executors.newCachedThreadPool(threadFactory), to manage worker threads.

[gemini-code-assist]

Similar to the JVM implementation, creating a new thread for each task can be inefficient, especially if tasks are numerous or short-lived. Using a shared thread pool (via ExecutorService) would be more performant by reusing threads.

[xerial]

Thanks for the review!

• Progress = Any: Addressed — the progress API was removed entirely in the refactor. Bodies publish progress / partial results / anything through any existing Rx primitive (RxVar / RxQueue / RxDeferred), so there's no Progress type on Task at all anymore. The comment is on a now-deleted file.
• Thread-per-task vs ExecutorService: Real concern. For the wvlet use cases (long-running DuckDB/Trino/Snowflake queries) per-task thread allocation is negligible relative to the work, so v1 stays simple. Updated the design doc to mark a shared scheduler as the right follow-up when a high-frequency caller appears.
• Re-interrupt on repeat cancel(): Considered, prefer keeping it one-shot. isCancelled is sticky, so well-behaved bodies will exit at their next checkpoint regardless. Re-interrupting on subsequent cancel() calls would arguably encourage bodies to swallow the first InterruptedException, which is exactly the failure mode we don't want.

[xerial]

/gemini review

[gemini-code-assist]

Code Review

This pull request implements wvlet.uni.control.Task, a cross-platform background work primitive supporting JVM, Scala Native, and Scala.js. It includes a Node.js implementation using worker_threads and SharedArrayBuffer to allow blocking await() calls. Reviewer feedback correctly identified a need to align the await() contract by ensuring InterruptedException is thrown upon cancellation and recommended utilizing Atomics.store for state transitions in the shared buffer to guarantee memory visibility across isolates.

[gemini-code-assist]

The current implementation of runBody in TaskImpl violates the Task.await() contract when a task is cancelled. The trait documentation states that await() should throw an InterruptedException on Cancelled, but runBody captures the original exception e and rethrows it directly (via failure in JvmTaskImpl/NativeTaskImpl). This means if a body throws a RuntimeException while cancelFlag is set, await() will throw that RuntimeException instead of an InterruptedException.

Additionally, awaitRx should be consistent with NodeWorkerTask (in taskCompat.scala), which always surfaces an InterruptedException for the Cancelled state. Wrapping the original exception as the cause of the InterruptedException preserves the diagnostic information while adhering to the API contract.

      case e: Throwable if cancelFlag.get() =>
        // Cancel was requested before the body finished — classify the throw as Cancelled even
        // if the exception type would otherwise be Failed (e.g. a JDBC InterruptedIOException).
        stateRef.set(Task.State.Cancelled)
        val ex = e match {
          case ie: InterruptedException => ie
          case _ =>
            val msg = if cancelReason == null || cancelReason.isEmpty then "Task cancelled" else cancelReason
            val ie  = new InterruptedException(msg)
            ie.initCause(e)
            ie
        }
        failure = ex
        completion.failure(ex)

[gemini-code-assist]

Add an atomicsStore helper to ensure memory visibility when updating the task state in the SharedArrayBuffer. While plain assignments to SharedArrayBuffer views are eventually visible, using Atomics.store provides the necessary memory barrier to ensure the parent thread (which is blocking on Atomics.wait) observes the update correctly.

  private def atomicsStore(view: Int32Array, index: Int, value: Int): Unit = js
    .Dynamic
    .global
    .Atomics
    .applyDynamic("store")(view, index, value)

[gemini-code-assist]

Use Atomics.store to update the state in the SharedArrayBuffer. This ensures that the write is visible to the parent thread before Atomics.notify is called.

      atomicsStore(stateView, 0, StateSuccess)

[gemini-code-assist]

Use Atomics.store to update the state in the SharedArrayBuffer when the task is cancelled.

        atomicsStore(stateView, 0, StateCancelled)

[gemini-code-assist]

Use Atomics.store to update the state in the SharedArrayBuffer when the task fails.

        atomicsStore(stateView, 0, StateFailed)

[gemini-code-assist]

Use Atomics.store to update the state in the SharedArrayBuffer when the worker exits unexpectedly.

          atomicsStore(stateView, 0, StateFailed)

[xerial]

Addressed in 2b9e6af:

• InterruptedException on Cancelled: TaskImpl.runBody now wraps any non-InterruptedException throw with new InterruptedException(reason) and attaches the original via initCause when the state is Cancelled, matching the await() contract. The body's original exception is preserved as the cause for debugging.
• Atomics.store for SAB state writes: added an atomicsStore helper and replaced all four stateView(0) = X assignments with it, giving the parent's Atomics.wait the memory barrier the spec actually guarantees visibility on.

All 33 tests still pass across JVM/JS/Native + domTest.