Signals

A signal is a named, type-safe background job. You define its input schema, handler function, and execution constraints. The runner picks it up, spawns an isolated child process, and manages retries, timeouts, and concurrency on your behalf.

signal(name)

Creates a named signal definition. The name must be unique across your application, start with a letter, and contain only letters, digits, hyphens, and underscores. Returns a builder with chainable methods.

import { signal, z } from "station-signal"; const sendEmail = signal("sendEmail")  .input(z.object({    to: z.string().email(),    subject: z.string(),    body: z.string(),  }))  .output(z.object({    messageId: z.string(),    sentAt: z.string().datetime(),  }))  .timeout(10_000)        // 10 seconds  .retries(3)             // up to 4 total attempts  .concurrency(5)         // max 5 concurrent sends  .onComplete(async (output, input) => {    console.log(`Email ${output.messageId} sent to ${input.to}`);  })  .run(async (input) => {    const result = await emailService.send(input);    return {      messageId: result.id,      sentAt: new Date().toISOString(),    };  });

Builder methods

.input(schema)

Sets the Zod schema for input validation. Every call to .trigger() validates the provided data against this schema before enqueuing. If validation fails, the run is rejected immediately and never enters the queue. The schema also drives TypeScript type inference: your handler receives the exact inferred type, and the compiler enforces it at build time.

const processOrder = signal("processOrder")  .input(z.object({    orderId: z.string().uuid(),    items: z.array(z.object({      sku: z.string(),      quantity: z.number().int().positive(),    })),  }))  .run(async (input) => {    // input is typed as { orderId: string; items: { sku: string; quantity: number }[] }    for (const item of input.items) {      await reserveInventory(item.sku, item.quantity);    }  });

If no input schema is provided, the signal accepts an empty object {} by default.

.output(schema)

Optional output schema. When provided, the handler's return value is validated against it. This is particularly useful when chaining signals inside broadcasts: the downstream signal's input type must match the upstream signal's output type. The validated output is JSON-serialized and stored on the run record, making it available to subscribers and the broadcast orchestrator.

const geocode = signal("geocode")  .input(z.object({ address: z.string() }))  .output(z.object({    lat: z.number(),    lng: z.number(),  }))  .run(async (input) => {    const coords = await geocodingApi.lookup(input.address);    return { lat: coords.latitude, lng: coords.longitude };  });

.timeout(ms)

Maximum execution time in milliseconds. If the handler exceeds this duration, the child process is killed with SIGTERM and the run is marked "failed" with a timeout error. Default: 300_000 (5 minutes). Set this lower for operations that should fail fast, such as HTTP requests or payment processing.

const healthCheck = signal("healthCheck")  .input(z.object({ url: z.string().url() }))  .timeout(5_000) // 5 seconds — fail fast if the service is down  .run(async (input) => {    const res = await fetch(input.url);    if (!res.ok) throw new Error(`Health check failed: ${res.status}`);  });

.retries(n)

Maximum retry attempts after the initial failure. Total execution attempts = 1 + n. Default: 0 (no retry). When a run fails and has remaining attempts, it is re-enqueued with "pending" status and an incremented attempt counter. The runner applies exponential backoff between retries (base delay configurable via retryBackoffMs on the runner). The retry delay doubles with each attempt: base * 2^(attempt - 1).

const syncInventory = signal("syncInventory")  .input(z.object({ warehouseId: z.string() }))  .retries(3)  // up to 4 total attempts (1 initial + 3 retries)  .run(async (input) => {    await externalApi.sync(input.warehouseId);  });

.concurrency(n)

Maximum concurrent executions of this specific signal. When the limit is reached, additional runs remain in the queue and are picked up on subsequent poll ticks as slots open. This is separate from the runner-level maxConcurrent setting, which caps total concurrent executions across all signal types. Use per-signal concurrency to rate-limit API calls or protect database-heavy operations.

const callStripeApi = signal("callStripeApi")  .input(z.object({ customerId: z.string(), amount: z.number() }))  .concurrency(3) // max 3 concurrent Stripe API calls  .retries(2)  .run(async (input) => {    await stripe.charges.create({      customer: input.customerId,      amount: input.amount,      currency: "usd",    });  });

.every(interval)

Makes the signal recurring. Accepts human-readable duration strings: "30s", "5m", "1h", "1d". After each completion, the runner automatically enqueues the next run. If a pending or running instance already exists for this signal, the scheduler skips that tick and advances the schedule to prevent overlapping executions.

const cleanupExpiredSessions = signal("cleanupExpiredSessions")  .every("15m")  .run(async () => {    const deleted = await db.sessions.deleteExpired();    console.log(`Cleaned up ${deleted} expired sessions`);  });

.withInput(data)

Default input data for recurring signals. Without this, recurring signals run with {} as input. The data must conform to the input schema if one is defined.

const dailyReport = signal("dailyReport")  .input(z.object({    reportType: z.enum(["summary", "detailed"]),    recipients: z.array(z.string().email()),  }))  .every("1d")  .withInput({    reportType: "summary",    recipients: ["ops@example.com", "team@example.com"],  })  .run(async (input) => {    const report = await generateReport(input.reportType);    await sendReport(report, input.recipients);  });

.step(name, fn)

Adds a named execution step. Steps run sequentially within the same child process. The first step receives the signal's input. Each subsequent step receives the return value of the previous step. Step completion events are emitted to subscribers as each step finishes, giving you granular progress tracking. When using steps, finalize with .build() instead of .run().

const processPayment = signal("processPayment")  .input(z.object({    orderId: z.string(),    amount: z.number(),    currency: z.string(),  }))  .step("validate", async (input) => {    const order = await db.orders.findById(input.orderId);    if (!order) throw new Error(`Order ${input.orderId} not found`);    return { order, amount: input.amount, currency: input.currency };  })  .step("charge", async (prev) => {    const charge = await paymentGateway.charge({      amount: prev.amount,      currency: prev.currency,    });    return { orderId: prev.order.id, chargeId: charge.id };  })  .step("confirm", async (prev) => {    await db.orders.update(prev.orderId, { chargeId: prev.chargeId, status: "paid" });    return { orderId: prev.orderId, chargeId: prev.chargeId };  })  .build();

.onComplete(fn)

Post-completion callback. Called with (output, input) after the handler finishes successfully. Runs in the same child process as the handler. If the callback throws, the run is still marked as completed but an onCompleteError event is emitted to subscribers. Use this for side effects like sending notifications, updating caches, or triggering downstream signals.

const generateInvoice = signal("generateInvoice")  .input(z.object({ orderId: z.string() }))  .output(z.object({ invoiceUrl: z.string().url() }))  .onComplete(async (output, input) => {    await notificationService.send({      channel: "email",      template: "invoice-ready",      data: { orderId: input.orderId, invoiceUrl: output.invoiceUrl },    });  })  .run(async (input) => {    const pdf = await renderInvoice(input.orderId);    const url = await storage.upload(pdf);    return { invoiceUrl: url };  });

.run(handler)

Sets the handler function and finalizes the signal. The handler receives the validated input and should return the output (or void). The handler runs in an isolated child process spawned by the runner, so it has its own memory space and cannot crash the runner.

const resizeImage = signal("resizeImage")  .input(z.object({    sourceUrl: z.string().url(),    width: z.number().int().positive(),    height: z.number().int().positive(),  }))  .run(async (input) => {    const image = await downloadImage(input.sourceUrl);    const resized = await sharp(image).resize(input.width, input.height).toBuffer();    await storage.upload(resized);  });

.build()

Finalizes a step-based signal. Use this instead of .run() when the signal is defined with .step() calls. Returns the same Signal object.

Signal instance

After calling .run() or .build(), you get a Signal object. This is the handle you use to trigger runs and the object you pass to the runner for registration.

.trigger(input)

Enqueues a run for execution. Validates the input against the schema first. Returns a Promise<string> resolving to the unique run ID. Throws SignalValidationError if the input fails validation. The run is written to the adapter with "pending" status and picked up by the runner on the next poll tick.

const runId = await sendEmail.trigger({  to: "user@example.com",  subject: "Order Confirmation",  body: "Your order has been placed.",}); console.log(`Enqueued run: ${runId}`);

SignalRunner

The runner is the long-running process that polls the adapter for due entries, spawns isolated child processes to execute signal handlers, and manages the full signal lifecycle: enqueue, dispatch, execute, retry, timeout, and completion. It also handles recurring signal scheduling, per-signal concurrency enforcement, and graceful shutdown on SIGINT/SIGTERM.

import { SignalRunner, ConsoleSubscriber } from "station-signal";import { SqliteAdapter } from "station-adapter-sqlite"; const runner = new SignalRunner({  signalsDir: "./src/signals",  adapter: new SqliteAdapter({ dbPath: "./data/signals.db" }),  subscribers: [new ConsoleSubscriber()],  pollIntervalMs: 1000,  maxConcurrent: 10,  retryBackoffMs: 2000,}); await runner.start();

Constructor options

Methods

SignalRunner.create(signalsDir, opts?)

Static factory method. Creates a runner with the given signals directory and a default ConsoleSubscriber if no subscribers are provided.

const runner = SignalRunner.create("./src/signals", {  adapter: new SqliteAdapter({ dbPath: "./data/signals.db" }),  maxConcurrent: 10,});

SignalQueueAdapter

The adapter interface defines storage operations for runs and steps. Implement this to use a custom storage backend. Two adapters ship with the framework: MemoryAdapter (built into station-signal) and SqliteAdapter (from station-adapter-sqlite).

Run

The Run interface represents a single execution of a signal. It is the primary record stored by the adapter.

Step

For multi-step signals, each step has its own record:

SignalSubscriber

All methods are optional. Implement only the events you need. Subscriber methods are called synchronously and should not throw. If a subscriber throws, the error is caught, logged, and does not affect signal execution.

The built-in ConsoleSubscriber logs all events to stdout with a [station-signal] prefix.

configure()

Sets a global default adapter. When the runner spawns a child process, that child may need to enqueue new runs (for example, if a signal's handler calls .trigger() on another signal). The child process reconstructs the adapter from serialized metadata and calls configure() automatically. You rarely need to call this directly; the runner does it for you in its constructor.

import { configure } from "station-signal";import { SqliteAdapter } from "station-adapter-sqlite"; configure({  adapter: new SqliteAdapter({ dbPath: "./data/signals.db" }),});

Re-exported Zod

import { signal, z } from "station-signal";

station-signal re-exports z from Zod v4. One import for schema definitions and signal definitions. No need to install Zod as a separate dependency.