sveltekit-tasks

Background task runner for SvelteKit with real-time progress streaming via Server-Sent Events (SSE).

Heads up: This is a single-process task manager. State is in-memory by default, with opt-in persistence (SQLite built in). It's designed as an easy drop-in for small self-hosted projects — not for production systems that need horizontal scaling. See Limitations for details.

Features

Server

• Background tasks with progress reporting, cancellation (AbortSignal) and per-task timeouts (timed_out status)
• run() for per-job tasks and wait() to await how a run ended
• Configurable history retention (maxHistory)
• Opt-in persistence via a pluggable adapter (SQLite built in) so task state survives restarts
• SSE endpoint with authorization and event replay

Client

• Reactive Svelte 5 TaskEventSource with auto-reconnect and connection status
• Change notifications (onChange), including what changed while the client was disconnected
• One shared connection for the whole app (provideTaskEvents / getTaskEvents)
• TaskItem component with snippet-based per-status rendering

Install

npm install sveltekit-tasks
pnpm add sveltekit-tasks
bun add sveltekit-tasks
deno add npm:sveltekit-tasks

Peer dependencies: svelte ^5.0.0, @sveltejs/kit ^2.0.0 || ^3.0.0

On SvelteKit 3, replace the $lib/... imports in the examples below with #lib/....

Quick Start

1. Define tasks on the server

// src/lib/server/my-tasks.ts
import { TaskManager } from "sveltekit-tasks/server";

export const tasks = new TaskManager();

// doWork() and handler stand in for your own code
tasks.register("import-data", async (ctx) => {
  for (let i = 0; i < 100; i++) {
    if (ctx.isCanceled()) return;
    ctx.progress("Processing...", i + 1, 100);
    await doWork(i);
  }
});

// Auto-cancel after 5 minutes
tasks.register("slow-sync", handler, { timeout: 300_000 });

During vite dev, editing this module, or one it imports, creates a second manager; see Keeping the manager across dev reloads.

2. Create the SSE and start/cancel endpoints

// src/routes/tasks/sse/+server.ts
import { tasks } from "$lib/server/my-tasks.js";

export const GET = tasks.createSSEHandler();

// src/routes/tasks/[action]/+server.ts — handles /tasks/start and /tasks/cancel
import { error, json } from "@sveltejs/kit";
import type { RequestHandler } from "./$types";
import { tasks } from "$lib/server/my-tasks.js";

export const POST: RequestHandler = async ({ params, request }) => {
  const { taskId } = await request.json();
  if (typeof taskId !== "string" || !taskId) error(400, "Invalid taskId");

  if (params.action === "start") tasks.start(taskId);
  else if (params.action === "cancel") tasks.cancel(taskId);
  else error(404, "Unknown action");

  return json({ ok: true });
};

With remote functions enabled, two commands that call tasks.start and tasks.cancel replace this endpoint, and the client passes them straight to TaskItem (onstart={startTask}).

3. Connect from the client

<!-- src/routes/+page.svelte -->
<script lang="ts">
  import { TaskEventSource, TaskItem } from "sveltekit-tasks/client";

  const taskEvents = new TaskEventSource("/tasks/sse");
  const taskList = $derived([...taskEvents.tasks.values()]);

  function post(action: "start" | "cancel", taskId: string) {
    fetch(`/tasks/${action}`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ taskId }),
    });
  }
</script>

{#each taskList as task (task.id)}
  <TaskItem {task} onstart={(id) => post("start", id)} oncancel={(id) => post("cancel", id)} />
{/each}

API

TaskManager (server)

import { TaskManager } from "sveltekit-tasks/server";
import { dev } from "$app/environment";

const tasks = new TaskManager({
  debug: dev, // log start() calls that do nothing (see below)
  maxHistory: 100, // keep the results of the 100 most recently finished tasks
  eventBufferSize: 1000, // buffer 1000 events for replay on reconnect
});

tasks.register(id, handler); // Register a task
tasks.register(id, handler, { timeout: 60_000 }); // ...that times out ("timed_out") after 60 s
tasks.register(id, handler, { ephemeral: true }); // ...that is unregistered when evicted by maxHistory
tasks.start(id); // Start a task (fire-and-forget)
tasks.run(id, handler, options); // Register if needed, then start: { id, started, done }
await tasks.wait(id, { signal }); // The state the run ends in
tasks.cancel(id); // Cancel a running task
tasks.getState(id); // Get one task's state
tasks.getAllStates(); // Get every task's state
tasks.subscribe(callback); // Subscribe to events (returns an unsubscribe function)

run() and wait() are covered in Per-job tasks.

The handler receives a TaskContext:

• ctx.progress(message, current?, total?) — report progress
• ctx.isCanceled() — check whether the task was canceled
• ctx.signal — an AbortSignal to pass to fetch() and the like

subscribe(callback) receives a TaskEvent. Narrow on event.type before reading event.state:

• { type: "update", taskId, state, eventId } — a task was registered, changed status or reported progress
• { type: "removed", taskId, eventId } — a task was unregistered

start() does nothing if the id is unknown or the task is already running, and cancel() does nothing if the task isn't running. Set debug: true to log the start() cases during development.

Cancellation is cooperative. cancel() aborts ctx.signal and marks the task "canceled" right away, but the handler keeps running until it checks ctx.isCanceled() or reacts to ctx.signal. If you cancel and immediately start again, the old handler keeps running alongside the new one until it notices; its late updates are ignored.

maxHistory

Limits how many finished tasks (completed, error, canceled or timed_out) keep their result. When the limit is exceeded, the oldest finished tasks (by lastRun) are evicted:

• Regular registered tasks are reset to "pending". The handler stays registered and the task can be started again.
• Tasks registered with { ephemeral: true } are unregistered entirely, handler included, and clients receive a "removed" message. Use this for per-job ids (import-${jobId}) so they don't pile up on a long-running server.

With persistence, evicted states are deleted from storage as well.

persistence

State is in-memory by default. Pass a PersistenceAdapter to restore task state after a server restart. The built-in sqliteAdapter has no dependencies and works with any synchronous SQLite driver — node:sqlite (Node 22.5+), better-sqlite3 or bun:sqlite:

import { DatabaseSync } from "node:sqlite"; // or: import Database from "better-sqlite3";
import { TaskManager, sqliteAdapter } from "sveltekit-tasks/server";

const db = new DatabaseSync("tasks.db");

export const tasks = new TaskManager({
  persistence: sqliteAdapter(db, { tableName: "sveltekit_tasks" }), // tableName is optional
});

// Restart automatically if the server stopped while this task was running
tasks.register("nightly-sync", handler, { restartInterrupted: true });

• Only status changes are persisted — progress updates are not.
• Persisted state is applied when a task is registered.
• Tasks that were "running" when the server stopped become "error" with error: "Interrupted by server restart" (exported as INTERRUPTED_ERROR), unless registered with restartInterrupted: true.
• States for ids that are never registered again stay in storage until maxHistory evicts them. That includes the "running" rows of per-job tasks a restart interrupted.
• Register restartInterrupted tasks at startup. Once any run in the new process ends, an interrupted row whose id isn't registered yet can be evicted, and that task won't restart.

Custom adapters implement three methods, each of which may be sync or async:

import type { PersistenceAdapter } from "sveltekit-tasks/server";

const adapter: PersistenceAdapter = {
  load: () => db.getAllTaskStates(), // called once, on construction
  save: (state) => db.upsertTaskState(state),
  delete: (taskId) => db.deleteTaskState(taskId),
};

With an async load(), persisted state is applied when it resolves — await tasks.ready if you need it first. Async writes are serialized in order; await tasks.flush() before a graceful shutdown to make sure they've landed. Write failures are logged and never affect in-memory state.

eventBufferSize

A client's first connection gets a full dump of every task's current state. When it reconnects, it sends the id of the last event it received — in the standard Last-Event-ID header, or the lastEventId query parameter that TaskEventSource uses — and gets either a replay of only the events it missed, or another dump.

eventBufferSize sets how many recent events are kept for replays. If the buffer still holds everything since the client's last event, the server replays; otherwise, or with 0 (the default), it sends a dump.

Event ids are tagged with a random epoch per TaskManager (<epoch>:<n>), so a client reconnecting after a server restart, or to a new manager, always gets a dump rather than a replay from the wrong one.

tasks.createSSEHandler(options?)

export const GET = tasks.createSSEHandler({
  authorize: (event) => event.locals.user?.isAdmin, // optional auth check
  heartbeatInterval: 30_000, // keepalive interval (ms)
});

Security: the SSE stream carries each task's full state, including the error message of failed runs, verbatim. If handler errors may contain sensitive details (connection strings, file paths), add an authorize check or sanitize errors before throwing.

TaskEventSource (client)

import { TaskEventSource } from "sveltekit-tasks/client";

const taskEvents = new TaskEventSource("/tasks/sse", {
  reconnectDelay: 1000, // initial reconnect delay (ms), default 1000
  maxReconnectDelay: 30_000, // max backoff delay (ms), default 30000
  maxRetries: 10, // max reconnect attempts, default 10
  onError: (event) => {}, // error callback
  initial: [], // task states from the server render
  onChange: (id, from, to) => {}, // called when a task appears, changes or disappears
});

initial and onChange are covered in Reacting to task changes.

• taskEvents.tasks — reactive SvelteMap<string, TaskState>, updated in real time from SSE messages
• taskEvents.status — reactive TaskEventSourceStatus: "connecting" (first attempt), "open", "reconnecting" (dropped, retry pending), "exhausted" (maxRetries consecutive failures) or "closed"
• taskEvents.connected / .exhausted / .closed — reactive boolean shorthands for status === "open" / "exhausted" / "closed"
• taskEvents.close() — close the connection and stop reconnecting (the task map is kept)
• taskEvents.reconnect() — (re)open the connection immediately and reset the retry counter

It reconnects with exponential backoff, asking for a replay when it can. The connection is closed automatically when the owning component is destroyed.

Use status rather than !connected for connection UI. On a fresh page load the connection is "connecting" until it opens, which is not an error state:

{#if taskEvents.status === "connecting"}
  <p>Connecting…</p>
{:else if taskEvents.status === "reconnecting"}
  <p>Connection lost, reconnecting…</p>
{:else if taskEvents.status === "exhausted"}
  <button onclick={() => taskEvents.reconnect()}>Reconnect</button>
{/if}

provideTaskEvents / getTaskEvents (client)

Share one TaskEventSource, and so one SSE connection, with every component instead of opening one per page. Call provideTaskEvents(url, options?) in the root layout; it takes the same options as TaskEventSource. Call getTaskEvents() in any component below it. Both must run during component initialization, and getTaskEvents() throws when no parent provided one.

<!-- src/routes/+layout.svelte -->
<script lang="ts">
  import { provideTaskEvents } from "sveltekit-tasks/client";
  let { children } = $props();
  provideTaskEvents("/tasks/sse");
</script>

{@render children()}

<!-- any page or component -->
<script lang="ts">
  import { getTaskEvents } from "sveltekit-tasks/client";
  const taskEvents = getTaskEvents();
</script>

TaskItem (client)

Renders a task based on its status. Each per-status snippet receives the narrowed state, so the completed snippet can read task.lastRun directly:

<TaskItem {task} onstart={handleStart} oncancel={handleCancel}>
  {#snippet running(task)}
    <p>{task.progress?.message}</p>
  {/snippet}
  {#snippet completed(task)}
    <p>Done at {new Date(task.lastRun).toLocaleTimeString()}</p>
  {/snippet}
  {#snippet error(task)}
    <p>Failed: {task.error}</p>
  {/snippet}
  {#snippet canceled(task)}...{/snippet}
  {#snippet timed_out(task)}...{/snippet}
  {#snippet pending(task)}...{/snippet}
</TaskItem>

Without snippets, it falls back to a default UI with Start/Cancel/Retry buttons.

Guides

Per-job tasks with run() and wait()

run(id, handler, options?) registers and starts a task in one call, which is what per-job ids (import-${jobId}) need. register() throws for an id that's already registered, but a second request for the same job often comes with new inputs:

• Not registered: registers handler with options, then starts it.
• Registered and not running: replaces the handler and options (options not passed again are cleared), then starts it.
• Already running: leaves the task and its handler alone; started is false.

It returns { id, started, done }, where done settles like wait().

wait(id, { signal }?) resolves with the state the run ends in ("completed", "error", "canceled" or "timed_out"), or at once with the current state when the task isn't running. It rejects for an unknown id, when the manager is disposed, or when signal aborts. Aborting stops the waiting, not the task, so cancel the task yourself if it should stop too:

// A parent task that runs each step of a job as its own task
tasks.run(`video:${jobId}`, async (ctx) => {
  for (const step of steps) {
    const stepId = `video:${jobId}:${step}`;
    tasks.run(stepId, (stepCtx) => runStep(jobId, step, stepCtx), { ephemeral: true });
    ctx.signal.addEventListener("abort", () => tasks.cancel(stepId), { once: true });

    const end = await tasks.wait(stepId, { signal: ctx.signal });
    if (end.status !== "completed") throw new Error(`${step} ended as ${end.status}`);
  }
});

Reacting to task changes

Pass onChange to run code when a task changes, for example to reload page data when a run ends. Pass the states the page was rendered with as initial, so the tasks render before the connection opens and changes since the server render are reported when it does:

// src/routes/+layout.server.ts
import { tasks } from "$lib/server/my-tasks.js";

export const load = () => ({ tasks: tasks.getAllStates() });

<!-- src/routes/+layout.svelte -->
<script lang="ts">
  import { untrack } from "svelte";
  import { invalidateAll } from "$app/navigation";
  import { provideTaskEvents } from "sveltekit-tasks/client";

  let { data, children } = $props();

  // Several tasks can change at once; reload once for all of them
  let reloading = false;
  function reload() {
    if (reloading) return;
    reloading = true;
    setTimeout(() => invalidateAll().finally(() => (reloading = false)), 100);
  }

  provideTaskEvents("/tasks/sse", {
    initial: untrack(() => data.tasks), // only the first render's states are wanted
    onChange: (id, from) => {
      if (from?.status === "running") reload(); // a run ended (or was replaced by a new one)
    },
  });
</script>

{@render children()}

onChange(id, from, to) receives the previous and the new TaskState:

• from is undefined for a new task, and to is undefined for a removed one.
• It fires when the status changes, and when a new run starts or ends with the same status (a new startedAt or lastRun). Progress updates don't fire it.
• Changes missed while disconnected are reported on reconnect, so there's no need to reload everything then. A replay delivers every missed change, but a dump only carries each task's latest state: a task that finished and started again shows up as running → running with a new startedAt. That's why the example checks from?.status === "running" rather than to?.status === "completed".
• Without initial, every task is reported as new on the first connect.
• Each task in a dump is its own message, so the callback runs once per changed task and can see taskEvents.tasks half-updated. Coalesce expensive work, as the example does.

Keeping the manager across dev reloads

During vite dev, keep a single TaskManager on globalThis:

// src/lib/server/my-tasks.ts
import { TaskManager } from "sveltekit-tasks/server";

const globalForTasks = globalThis as typeof globalThis & { __tasks?: TaskManager };

export const tasks = (globalForTasks.__tasks ??= createTasks());

function createTasks() {
  const tasks = new TaskManager();
  tasks.register("import-data", importData);
  return tasks;
}

When you edit a server module, Vite evaluates it again, and a module-level new TaskManager() creates a second manager:

• Tasks still running on the old manager vanish from the UI on the next reconnect or page load.
• With persistence, the new manager loads those running tasks from storage and marks them interrupted — or, with restartInterrupted, starts a second run alongside the first.

Everything inside createTasks() runs once per dev-server process, so edited handlers, new register() calls and changed options need a restart (until then, start() on a new id does nothing). run() replaces the handler on each call while the task is idle, so it picks up edits. In production the module is evaluated once, so the pattern changes nothing.

Types

TaskState is a discriminated union on status — narrow via task.status === "running" etc. to access status-specific fields:

import type {
  TaskStatus, // "pending" | "running" | "completed" | "error" | "canceled" | "timed_out"
  TaskProgress, // { message, current?, total? }
  TaskState, // discriminated union (see below)
  TaskContext, // { progress(), isCanceled(), signal }
} from "sveltekit-tasks";

import type {
  TaskEvent, // TaskUpdateEvent | TaskRemovedEvent — what subscribe() receives
  PersistenceAdapter,
} from "sveltekit-tasks/server";

| Status | Fields | | ------------- | ------------------------------- | | "pending" | id | | "running" | id, startedAt?, progress? | | "completed" | id, lastRun | | "error" | id, lastRun, error | | "canceled" | id, lastRun | | "timed_out" | id, lastRun |

Limitations

• In-memory by default — without a persistence adapter, a server restart loses all state. Even with one, running handlers are not resumed mid-run: interrupted tasks are marked as failed or restarted from scratch.
• Single-instance — state is held in a JS Map. In multi-process or multi-server deployments, tasks on one instance are not visible to SSE connections on another.
• No concurrency control — all registered tasks can run simultaneously. Implement your own limiter if needed.
• No progress throttling — every ctx.progress() call emits an SSE message. If your task reports progress in a tight loop, consider adding your own debounce/throttle to avoid flooding clients.

Future Additions

These are not currently planned but could be added in the future:

• More persistence adapters — e.g. Postgres, Redis.
• Task metadata — a meta object on register()/run() sent with the state, so clients can show labels and links without encoding them in the id.
• Horizontal scaling — shared state across multiple server instances via an adapter.
• Concurrency control — maxConcurrent option to limit how many tasks run simultaneously, with a queue for excess.
• Task scheduling / queuing — delayed execution, priority queues, cron-like scheduling.

License

MIT