Background jobs

Inside a view, async submits work to the application's worker executor, separate from the native UI thread. It can be called from on_mount, an event handler, or a method invoked on the mounted view's owning context. For example, a button can start a file checksum while the window remains responsive:

require "digest/sha256"

class ChecksumView
  include CryKit::View

  reactive path : String = ""
  reactive status : String = "Choose a file path."
  reactive busy : Bool = false

  private def calculate
    path = reactive.path.dup
    reactive.busy = true

    async do |job|
      begin
        checksum = Digest::SHA256.hexdigest(File.read(path))
        job.notify do
          reactive.status = checksum
          reactive.busy = false
        end
      rescue ex
        message = ex.message || "Could not read the file."
        job.notify do
          reactive.status = message
          reactive.busy = false
        end
      end
    end
  end

  template "checksum"
end

Save checksum.ecr beside the Crystal source file:

<input binds=<%= reactive.path %> placeholder="File path">
<button disabled=<%= reactive.busy %> on_click=<%~ -> { calculate } %>>Calculate checksum</button>
<p role="status"><%= reactive.busy ? "Calculating…" : reactive.status %></p>

Named jobs and streams

Declare repeated operations at class scope to keep their handles out of view state. A named job has a stable CryKit::JobController accessor:

job save do |worker|
  if worker.sleep(500.milliseconds)
    worker.notify { flush_pending }
  end
end

private def schedule_save : Nil
  save.restart
end

private def cancel_save : Nil
  save.cancel
end

save.start and save.restart both start a fresh run of the declared body, cancelling the previous job and its queued deliveries first. The declaration does not start work by itself. save.cancel is safe before the first run and after unmount. The notes editor uses this declaration to debounce saves without an optional job field.

For inputs captured when starting work, omit the declaration body and pass a block to start:

job checksum

private def calculate : Nil
  path = reactive.path.dup
  checksum.start do |worker|
    result = Digest::SHA256.hexdigest(File.read(path))
    worker.notify { reactive.status = result }
  end
end

Starting with a block uses that block for the new run. Capture mutable inputs on the UI context before starting; a declaration does not copy values or make worker access to view state safe. Declared job bodies execute on the same worker executor as async. Deliver reactive state changes through worker.notify so cancelling the job also discards its pending deliveries.

Declare a stream with its payload type and UI delivery handler:

stream scan : DirectoryHasher::Progress do |snapshot|
  reactive.progress = snapshot
end

private def start_scan : Nil
  scan.start(DirectoryHasher.new(reactive.path.dup))
end

private def cancel_scan : Nil
  scan.close
end

The stable accessor returns a CryKit::StreamController(T). Each start(source) closes the previous stream, creates a fresh view-owned stream, and subscribes the declared handler. A declaration without a handler accepts scan.start(source) { |snapshot| ... } instead. close is safe when idle; pause removes the current listener and its queued deliveries while source iteration continues, and resume subscribes to future values. Neither resume nor a finite source's completion restarts source iteration. Call start with a fresh source to begin another run.

Add from: to create and subscribe a stream automatically on each mount:

stream updates : MockMarketFeed::Tick, from: @source do |tick|
  reactive.latest = tick
end

The source expression is evaluated after the view owns a mount scope and before its on_mount hook. It requires a declaration handler. Standalone rendering does not evaluate it or start work. All named operations release their active handles on unmount, including failed mounting; remounting never revives old jobs or deliveries. Start, restart, pause, resume, and close controllers from their view's owning context, just like other mutable view state. Starting work requires a mounted view with an update-capable host.

The imperative async { |job| ... } and stream(source) APIs remain available for one-off work and independent subscriptions.

Delivering updates

Native listeners created in a mounted view's mount hook or UI callbacks run in that view's scope. Click handlers, native callbacks, and mount hooks can assign reactive fields directly; those writes schedule the affected bindings automatically.

async and job.every execute on a worker. Use job.notify { ... } to hand reactive state changes back to the UI context, serialize them with UI callbacks, and discard pending deliveries when the job is cancelled. Assigning a reactive field does not move execution to the UI thread. Fields without reactive rendering dependencies do not request rendering. Notification does not complete the worker: it can continue computing, submit another update, or return.

DOM effects such as dom.focus("note-title") schedule their own delivery and can be called directly from the worker. Effects requested by a job retain that job's cancellation ownership, including requests sent to another view. See DOM effects.

The initiating click renders the busy state through the normal event cycle. job.notify submits a short state-change block for a UI update slot. Reactive writes in that block update their readers. It returns immediately, without waiting for execution or DOM acknowledgment. It returns true when accepted and false when the queue is full or the job, scope, or host has closed. Each root mount retains at most 1,024 pending callbacks (shared by its children). Overload rejects the newest attempt without evicting accepted work. Producers must check the result and choose to throttle, retry while active, or discard an optional progress sample. Avoid busy retries. Every accepted, still-valid block runs in submission order. Large bursts are processed in chunks so other UI events can run between them. Individual blocks must remain short: queued work still runs on the UI context.

Capture inputs before starting work. Worker blocks must not directly read or write mutable view state; use local results and job.notify for state changes. Captured objects are not automatically copied or synchronized. Expensive work inside job.notify or a template reader still blocks the UI. Uncaught worker exceptions are logged; an exception in an update block or template reader fails that mount, and previous state changes are not rolled back.

Cancellation

async returns a CryKit::Job. Call job.cancel to discard its pending updates, and check job.cancelled? between units of long-running work. Cancellation is cooperative: it prevents pending jobs from starting and suppresses future updates, but does not forcibly interrupt running computation or blocking I/O. Removing a view, navigating away, or closing the application cancels its jobs. Cancellation does not itself render an application-specific status message.

Use job.sleep(duration : Time::Span) for a cancellation-aware pause. It returns true when the duration elapses while the job remains active, or false if the job is cancelled before or during the pause. Closing its scope, update queue, application lifetime, or executor also ends the pause.

The pause checks cancellation roughly every 10 milliseconds, subject to scheduling. Negative durations raise ArgumentError; zero yields once when the job is active, then checks cancellation again.

For repeated work, job.every(interval : Time::Span, &) waits before the first callback, then runs callbacks sequentially on the worker, waiting the full interval after each callback finishes. With a reactive ticks : Int32 field:

async do |job|
  job.every(1.second) do |worker|
    worker.notify { reactive.ticks += 1 }
  end
end

The every block receives its job handle so state changes can be delivered with worker.notify. Cancellation discards pending deliveries, stops waits and future callbacks, but cannot interrupt a running callback. Negative intervals raise ArgumentError; zero yields once per tick while active. every returns nil when it stops.

Application jobs and channels

Use app.async for work that should continue across navigation and window closure while the application remains running. It returns a CryKit::AppJob with cancel, cancelled?, sleep, and every. Application shutdown cancels these jobs automatically, including work that has not started. They use the same worker executor and cooperative cancellation rules as view jobs, and can be submitted before app.run.

An application job publishes values with app.broadcast_to(channel, value):

time_job = app.async do |job|
  job.every(1.second) { app.broadcast_to(:current_time, Time.local) }
end

# Stop this publisher early when needed:
# time_job.cancel

Views declare channel dependencies, and the mount registers them only while each view is mounted. Broadcasts queue callbacks only for live mounted consumers, on each consumer's UI context, where they can change view state directly. Unmount removes those dependencies and suppresses pending delivery. Application jobs can keep publishing after every listening view has unmounted. Application jobs have no job.notify method; app.broadcast_to publishes a value through the application's channel source.

app.broadcast_to is nonblocking and returns true when all active subscriptions accept delivery, including when there are no listeners. It returns false if the application has shut down or any consumer's bounded queue rejects delivery. Other consumers may already have accepted the value, so retrying the whole publication can duplicate their delivery. An optional clock sample can be discarded when rejected. Accepted callbacks run in their consumer queue's submission order. Cancelling the publisher does not retract values already accepted by subscribers.

Each channel Symbol fixes its payload type on first publication or subscription. A different payload type raises ArgumentError, even after its listeners leave. Values are not replayed to later subscribers or copied: publish immutable values or independent snapshots. Worker code must not read or mutate shared UI state without synchronization.

The demo's clock publisher sends Time values from one application job. Its Counter view subscribes only while mounted. ClockState still owns Pause / Resume, and the view reads that state in its UI callback. Application state and channel delivery can be adopted separately; existing state objects remain ordinary Crystal code.

File-hashing example

The demo's File hashes view (/files) consumes DirectoryHasher, a plain Crystal Enumerable(DirectoryHasher::Progress) that yields snapshots while discovering files and hashing them with SHA-256. The view declares its handler once:

stream scan : DirectoryHasher::Progress do |snapshot|
  reactive.progress = snapshot
end

Its Start handler calls scan.start(DirectoryHasher.new(reactive.scan_root)) after validating the path and setting the discovering status. The hasher has no CryKit dependency. The stream reads it on a worker and delivers snapshots on the UI context, where the view updates its progress and results.

Cancel calls scan.close to discard pending deliveries and sets the view's cancelled status immediately. Scanning stops at its next progress yield. The view's mount scope also closes the stream when the page is left.

External service example

The demo's Market view (/market) consumes a mock market feed, an ordinary Crystal Enumerable(MockMarketFeed::Tick). Its each method yields fictional BTC, ETH, and SOL snapshots at irregular intervals, without network access. It has no CryKit dependency, worker context, subscription registry, or cancellation tokens. A real source can yield whenever incoming data is available; CryKit does not poll the source or impose an update interval.

The view declares the mounted source and its handler:

stream updates : MockMarketFeed::Tick, from: @source do |tick|
  reactive.latest = tick
  reactive.received += 1
  mutate reactive.recent do |recent|
    recent.unshift(tick)
    recent.pop if recent.size > 8
  end
end

The declaration creates a stream and its listener on mount. The listener starts reading the source on the application's worker executor. Handlers run on the view's UI context and assign reactive fields directly; those writes schedule their rendering readers. Scheduling and cancellation stay inside CryKit. Sources must yield immutable values or independent snapshots: CryKit does not copy yielded objects. Source iteration slows when the UI delivery queue is full, preserving values and their order for active subscribers.

Pause updates calls updates.pause, removing the listener and discarding its queued deliveries while source iteration continues. Resume updates calls updates.resume, receiving future values; it does not replay missed snapshots or restart the source. Finite sources are read once, and completion preserves already queued deliveries.

For independent listeners, the imperative API remains available:

feed = stream(@source)
unsubscribe = feed.subscribe { |tick| reactive.latest = tick }
# Stop this listener early:
unsubscribe.call

CryKit closes view-owned streams automatically when their mount ends. close is idempotent and immediately suppresses all pending and future deliveries. Reading stops cooperatively when the source next yields; closing a stream does not interrupt blocking I/O or arbitrary source code. A source that owns a socket or another resource should release it in its own ensure block. Subscribing to a closed raw CryKit::Stream(T) raises an error; a named controller can start a fresh stream with start(source).

Ownership and executor configuration

Each view job belongs to the view's current mounting. A removed child cannot update its former parent, and remounting an instance does not revive its old jobs. Update scopes expire before on_unmount. Results arriving before browser readiness are retained until the handshake completes. A view instance can have only one live mounting; create separate instances for separate windows. Standalone HTML and unhosted mounts cannot start mounted jobs. Application jobs belong to the application lifetime and do not require a mounted view.

Configure worker parallelism with CryKit::Application.new(worker_parallelism: 4). The default is max(1, System.cpu_count - 1). The default executor also limits outstanding jobs, including suspended jobs, to 1024. Set worker_capacity: to change this admission limit. Submission at capacity raises CryKit::Execution::WorkerExecutor::CapacityError before creating a fiber or retaining the work. Submission never waits on the UI context; handle this error and retry after existing work completes, or reuse a job/controller instead of starting a job for every input event. Applications can inject a CryKit::JobExecutor with executor:; implementations must enqueue work, implement closed? and close, and never execute it on the submitting context. Shutdown stops accepting work and invalidates updates without joining arbitrary running jobs. Navigation and shutdown do not wait for a job to finish.

The view example above requires CryKit::View to be loaded and the view to be mounted by an application. See Views for lifecycle rules and Runtime protocol for the shared delivery queue.