Run a Temporal worker

How-to. The lesson that fronts this recipe: The same application, a second runtime. Boot a Temporal worker under the kernel: activities built from your own services, the connection as a resource of the graph, and a drain that releases the kernel at the kernel's deadline. For the package's full surface, see @btravstack/temporal-worker; for why the drain needs a package of its own, see Draining, in three beats.

You have a temporal-contract contract and the use cases its activities call. What you write is the activities record and a composition root; the Worker's create / run / shutdown, the unit per activity attempt and the deadline race are the package's. Everything below is lifted from examples/order-temporal-worker.

Recipe

1. Implement the activities with TemporalActivities(contract)({ inject?, unit?, sync }) — a record shaped like the contract, closing over the services inject names.
2. Compose with TemporalModule(name)({ contract, activities, workflows, imports, needs }).
3. await runMain(OrderTemporalWorker).
4. Set TEMPORAL_ADDRESS / TEMPORAL_NAMESPACE in the deployment; keep forceAfter at or below the kernel's drainTimeoutMs.

Step 1 — the activities, as a provider

TemporalActivities(orderContract) is the builder on the starter's own activities port, typed for the contract (its service the record declareActivitiesHandler takes) — no class, no name: a worker serves one activities record — so the next call is { inject?, unit?, sync }, the same arm all three starters take. An activity is a closure over its provider's services — the record it receives at call time carries the invocation's own values (input, errors), never a service to resolve — and each mapErrCases names every domain error the contract declares:

A single record covers every workflow the contract declares — orderContract has three, fulfillOrder, chargeOrder and the scheduled sweepStaleOrders (whose one activity is the example's withdrawStale, see Run something on a schedule):

import {
  OrderRepository,
  PaymentService,
  PlaceOrder,
  ShippingService,
  StockService,
} from "@btravstack/example-order-application";
import { TenantId } from "@btravstack/example-order-domain";
import { orderContract } from "@btravstack/example-order-temporal-contract";
import { TemporalActivities } from "@btravstack/temporal-worker";
import { Context } from "@temporalio/activity";
import { P } from "unthrown";

export const orderActivities = TemporalActivities(orderContract)({
  inject: {
    stock: StockService,
    shipping: ShippingService,
    payments: PaymentService,
  },
  // The tenant-bound use cases, read off the fork the worker opens per
  // attempt — see the unit module in Step 2.
  unit: { place: PlaceOrder, repository: OrderRepository },
  sync: ({ stock, shipping, payments }) => ({
    fulfillOrder: {
      place: ({ errors, context, input }) =>
        context.unit.place
          .execute(input.orderId, input.quantity, { operation: input.operationId ?? Context.current().info.workflowExecution!.runId })
          .map((order) => ({ id: order.id, quantity: order.quantity }))
          .mapErrCases((matcher) =>
            matcher
              .with(P.tag("InvalidQuantity"), (error) =>
                errors.InvalidQuantity({ id: error.id }),
              )
              .with(P.tag("InvalidOrderId"), (error) =>
                errors.InvalidOrderId({ id: error.id }),
              )
              .with(P.tag("DuplicateOrder"), (error) =>
                errors.OrderAlreadyPlaced({ id: error.id }),
              ),
          ),
      reserveStock: ({ errors, input }) =>
        stock
          .reserve(input.orderId, input.quantity)
          .mapErrCases((matcher) =>
            matcher.with(P.tag("OutOfStock"), (error) =>
              errors.OutOfStock({ id: error.id }),
            ),
          ),
      arrangeShipping: ({ errors, input }) =>
        shipping
          .arrange(input.orderId)
          .mapErrCases((matcher) =>
            matcher.with(P.tag("ShippingUnavailable"), (error) =>
              errors.ShippingUnavailable({ id: error.id }),
            ),
          ),
      releaseStock: ({ input }) => stock.release(input.orderId),
      cancelPlacement: ({ context, input }) =>
        context.unit.repository
          .remove(input.orderId, { operation: input.operationId ?? Context.current().info.workflowExecution!.runId })
          .recoverErrCases((matcher) =>
            matcher.with(P.tag("OrderNotFound"), () => undefined),
          ),
    },
    chargeOrder: {
      authorizePayment: ({ errors, idempotencyKey, input }) =>
        payments
          .authorize(input.orderId, input.amount, idempotencyKey)
          .map((authorizationId) => ({ authorizationId }))
          .mapErrCases((matcher) =>
            matcher.with(P.tag("PaymentDeclined"), (error) =>
              errors.PaymentDeclined({ id: error.id }),
            ),
          ),
      capturePayment: ({ idempotencyKey, input }) =>
        payments.capture(input.authorizationId, idempotencyKey),
      refundPayment: ({ idempotencyKey, input }) =>
        payments.refund(input.authorizationId, idempotencyKey),
    },
    sweepStaleOrders: {
      withdrawStaleOrders: ({ context, input }) =>
        withdrawStale(
          context.unit.repository,
          new Date(input.placedBefore),
          currentUnit()?.signal,
        ),
    },
  }),
});

The package maps nothing further: declareActivitiesHandler already turns a declared contract error into a nonRetryable ApplicationFailure the workflow branches on, and leaves anything unmodeled to Temporal's retry policy. Naming a failure here is also what tells Temporal to stop retrying it.

input.tenantId is the application's own, declared on every workflow and activity input by the contract — so Temporal persists it in the event history and a replay reconstructs it, and the package reads nothing about tenancy. The worker seeds the per-attempt fork with that very input on ActivityInput(contract), and the unit module in Step 2 turns it into a Tenant once, claiming examples/order-domain's brand there rather than at each activity. So PlaceOrder and OrderRepository arrive on context.unit already bound, and no activity passes a tenant. cancelPlacement absorbs OrderNotFound on purpose — a compensation Temporal may re-run has to answer the same both times. Its operationId is the placing workflow's run id, checked in the repository's DELETE so a late compensation leaves another run's replacement alone. A task scheduled by an older workflow may omit it; the activity then reads that same run id from Temporal's native metadata. A historical row with no stored operation is left alone because ownership cannot be verified.

Step 2 — the composition root

import { Logger } from "@btravstack/core";
import { Module, Provider } from "@btravstack/di";
import {
  OrderApplicationModule,
  Tenant,
} from "@btravstack/example-order-application";
import { TenantId } from "@btravstack/example-order-domain";
import {
  OrderDatabase,
  OrderPersistenceModule,
  OrderTenantPersistence,
} from "@btravstack/example-order-infrastructure";
import { orderContract } from "@btravstack/example-order-temporal-contract";
import { observability } from "@btravstack/observability";
import { otel } from "@btravstack/observability/otel";
import { ActivityInput, TemporalModule } from "@btravstack/temporal-worker";
import { workflowsPathFromURL } from "@temporal-contract/worker/worker";

import { orderActivities } from "./activities.js";
import { BillingModule } from "./billing.js";
import { FulfillmentModule } from "./fulfillment.js";

// Forked per attempt and seeded with the validated input, so the tenant the
// contract carries is claimed once and the orders vertical is composed over
// it. `OrderTenantPersistence` reads `OrderDatabase` out of the application
// scope rather than importing the database module: one client per process,
// not one per attempt.
export const ActivityUnitModule = Module("ActivityUnit")({
  imports: [OrderTenantPersistence, OrderApplicationModule],
  provides: [
    Provider(Tenant)({
      inject: { input: ActivityInput(orderContract) },
      sync: ({ input }) => TenantId(input.tenantId),
    }),
  ],
  exports: [Tenant, PlaceOrder, OrderRepository],
});

export const OrderTemporalWorker = TemporalModule("OrderTemporalWorker")({
  contract: orderContract,
  activities: orderActivities,
  workflows: {
    workflowsPath: workflowsPathFromURL(import.meta.url, "./workflows.js"),
  },
  imports: [
    OrderPersistenceModule,
    FulfillmentModule,
    BillingModule,
    observability(),
    // `BillingModule` counts its authorizations through `Meter` — otel()
    // is what supplies it, beside the `Logger` observability() does.
    otel(),
  ],
  unit: { activity: ActivityUnitModule },
});

TemporalModule is Module(name)({...}) plus the starter's fields: it imports the Temporal starter for contract and workflows, provides the activities and exports TemporalRuntime. observability() is the other starter in the list — the Logger the use case, the fulfillment services and the billing stand-in write to, bound from LOG_LEVEL, JSON per line on stdout, every line carrying the activity attempt's own trace id. The starter's runtime provider depends on its activities port through di, so a root whose imports do not cover what the provider declared (FulfillmentModule and BillingModule here — chargeOrder's PaymentService comes from the latter) is refused at start — the Needs channel failing to assign against Env | Scope, which names the port, not di's UNSATISFIED DEPENDENCIES dependency gate; a root with no starter is refused against "NO RUNTIME — the module exports no port declared over RuntimePort". activities is typed against the module's own contract: a provider built for another contract is refused at the call.

workflows is a WorkflowSource: { workflowsPath } for a process that lets Temporal bundle the module, { workflowBundle } for a spec that built one and memoised it.

Step 3 — main.ts

import { runMain } from "@btravstack/core";

import { OrderTemporalWorker } from "./module.js";

await runMain(OrderTemporalWorker);

Configuration and options

The starter provides three ports: TemporalRuntime, TemporalConfig ({ address, namespace }) and TemporalConnection — the NativeConnection as a resource, opened with the scope and closed on every exit path, startup failure included.

Variable / option	Default	Notes
TEMPORAL_ADDRESS / address	127.0.0.1:7233	a blank value is a ConfigInvalid, exit 78
TEMPORAL_NAMESPACE / namespace	default	likewise
gracePeriod	10 seconds	Temporal's shutdownGraceTime
forceAfter	15 seconds	Temporal's shutdownForceTime — keep it ≤ drainTimeoutMs

address / namespace pin a field instead of reading it — explicit beats environment beats default, per field:

export const Pinned = TemporalModule("OrderTemporalWorkerLocal")({
  contract: orderContract,
  activities: orderActivities,
  workflows: {
    workflowsPath: workflowsPathFromURL(import.meta.url, "./workflows.js"),
  },
  address: "127.0.0.1:7233",
  gracePeriod: "5 seconds",
  forceAfter: "15 seconds",
  imports: [
    OrderPersistenceModule,
    FulfillmentModule,
    BillingModule,
    observability(),
    // `BillingModule` counts its authorizations through `Meter` — otel()
    // is what supplies it, beside the `Logger` observability() does.
    otel(),
  ],
  unit: { activity: ActivityUnitModule },
});

A service that will not answer is a modeled TemporalUnreachable{ address, cause } — a startup Err, so runMain exits 1: an operator can act on it, and it is not the 70 a defect earns. A contract the activities record cannot satisfy (an implementation the contract never declared, one it declares and finds missing) is RuntimeStartFailed, also 1.

Read what the worker published

Once polling, the runtime publishes TemporalInfo — { taskQueue, namespace } — on Serving.info:

const app = start(OrderTemporalWorker, {
  env: {
    TEMPORAL_ADDRESS: "127.0.0.1:7233",
    DATABASE_URL: "postgres://orders@localhost:5432/orders",
  },
  signals: false,
  probes: false,
});
const info = (await app.runtimeInfo()).get(); // TemporalInfo | undefined

The specs boot the same TemporalModule sugar with env: { TEMPORAL_ADDRESS, TEMPORAL_NAMESPACE } pointing at the Temporal server the whole repository shares — a namespace of the spec file's own, a per-test task queue and a memoised bundle.

The drain, and the one surprising behaviour

Serving.drain(signal) calls worker.shutdown() — polling stops at once, in-flight activities run to completion — then waits on run() raced against the kernel's deadline signal. run() settles on Temporal's own shutdownForceTime, not the kernel's drainTimeoutMs, so without the race an activity that never finishes would hold Serving.stop well past the kernel's deadline.

When the deadline wins

The runtime returns and the worker is still alive: @temporalio/worker exposes no public forced shutdown, so "stop waiting" is the only escalation. The kernel gets its thread back on time, reports the activity abandoned (exit 2 under runMain), and the worker keeps winding down on Temporal's clock. Connection release waits for the worker to finish before closing it. The kernel gives that finalizer its existing stopTimeoutMs; if that expires, the report also names abandonedAt: "stop" and cleanup continues. Unexpected close failures remain in the report's live teardownErrors array, including failures arriving after exit, but no lifecycle event follows exited.

Keep forceAfter at or below drainTimeoutMs (default 20_000). It matters most on the stop()-only path — no signal, no kernel deadline — where Temporal's clock alone decides when Serving.stop returns. The package cannot do this for you: drainTimeoutMs is a StartOptions field the runtime never sees.

Activities that differ in scope: a piece's own unit module

unit.activity is one module for the whole worker, forked around every attempt. That breaks down when the activities do not share a scope — a scheduled job fanned out per tenant, say: the parent's activity lists the tenants and has no tenant in its input, while each child's activities act for one. A Tenant derived from the input in the worker-wide module would be built for the listing activity too.

A piece names its own module with fork. Its attempts fork that module — seeded with the validated input on ActivityInput(contract), like unit.activity — and every other piece keeps the worker-wide one:

const activityOptions = { startToCloseTimeout: "30 seconds" } as const;

const contract = defineContract({
  taskQueue: "invoicing",
  workflows: {
    invoiceAll: defineWorkflow({
      input: z.object({}),
      output: z.number(),
      startPolicy: "allow-duplicate",
      activities: {
        listTenants: defineActivity({
          input: z.object({}),
          output: z.array(z.string()),
          activityOptions,
        }),
      },
    }),
    invoiceTenant: defineWorkflow({
      input: z.object({ tenantId: z.string() }),
      output: z.void(),
      startPolicy: "allow-duplicate",
      activities: {
        issueInvoices: defineActivity({
          input: z.object({ tenantId: z.string() }),
          output: z.void(),
          activityOptions,
        }),
      },
    }),
  },
});

const Input = ActivityInput(contract);

// Built per attempt of the pieces that name it, and only those.
const TenantUnit = Module("TenantUnit")({
  provides: [
    Provider(Tenant)({
      inject: { input: Input },
      sync: ({ input }) => ("tenantId" in input ? input.tenantId : ""),
    }),
  ],
  exports: [Tenant],
});

const listing = TemporalWorkflowActivities(
  contract,
  "invoiceAll",
)({
  inject: { tenants: Tenants },
  sync: ({ tenants }) => ({ listTenants: () => OkAsync([...tenants.list()]) }),
});

const perTenant = TemporalWorkflowActivities(
  contract,
  "invoiceTenant",
)({
  unit: { tenant: Tenant },
  fork: TenantUnit,
  sync: () => ({ issueInvoices: ({ context }) => (void context.unit.tenant, OkAsync()) }),
});

export const InvoicingWorker = TemporalModule("InvoicingWorker")({
  contract,
  activities: TemporalActivities(contract)([listing, perTenant]),
  workflows: { workflowsPath: "./workflows.js" },
  provides: [
    listing,
    perTenant,
    Provider(Tenants)({ inject: {}, value: { list: () => ["acme", "globex"] } }),
  ],
});

The ports unit declares are checked against the piece's own module, at the piece's call — UNIT DOES NOT PROVIDE there, rather than at the root — and the module's own unmet needs reach start exactly as unit.activity's do. A spec overrides inside it by the module's name: overridden(root, [], { unit: { TenantUnit: [stub] } }).

The unit boundary

One unit per activity attempt, not per execution: UnitMeta.id is Temporal's base64 task token — unique per attempt by Temporal's guarantee — and traceId is the workflow id, the correlation id minted outside this process and stable across every retry. An adapter reads either from currentUnit(); the middleware injects nothing into the activity itself.

Honouring the drain deadline

Because the middleware injects nothing, currentUnit()?.signal is the only route to the unit's AbortSignal from inside an activity — there is no parameter to receive one through, and adding a context the contract does not type was the alternative. It is aborted at the kernel's drainTimeoutMs:

arrange: (orderId) =>
  currentUnit()?.signal.aborted === true
    ? fromSafePromise(
        Promise.reject(
          new Error(
            `the drain deadline passed before shipping for ${orderId} was arranged`,
          ),
        ),
      )
    : (logger.info("arranged shipping", { orderId }), OkAsync()),

Failing as a defect is deliberate: the platform retries that attempt on another worker, which is what "we ran out of time here" means. A modeled contract error — ShippingUnavailable — is a permanent no and would be wrong.

Temporal's own Context.current().cancellationSignal is a different clock: it fires on a workflow-side cancellation, and on worker shutdown after shutdownGraceTime. Honour both; neither stands in for the other.

See also

• @btravstack/temporal-worker — options, ports, TemporalInfo, WorkflowSource.
• Split a worker into slices — several workflows, one activities record per workflow, composed at the root.
• Order Temporal worker — the saga these samples come from.
• Tune the drain for Kubernetes — drainTimeoutMs against terminationGracePeriodSeconds.
• Configure from the environment — how TEMPORAL_* are bound and pinned.