@btravstack/temporal-worker

The worker half of Temporal for @btravstack/core — starting workflows is a client's job, and belongs to @temporal-contract/client. One module providing the runtime, its configuration and its connection; one unit per activity attempt; and a drain that releases the kernel at the kernel's deadline, not Temporal's shutdownForceTime.

📖 Documentation · Reference · API Reference

pnpm add @btravstack/temporal-worker @btravstack/core @btravstack/config @btravstack/di unthrown \
  @temporalio/worker @temporalio/activity @temporalio/common \
  @temporal-contract/worker@^8.0.0-beta.7 @temporal-contract/contract@^8.0.0-beta.7

All nine are peer dependencies — install them. Node >=22.22. This package is ESM-only.

A worked example

import { runMain, Logger } from "@btravstack/core";
import { Provider } from "@btravstack/di";
import {
  ActivityInput,
  TemporalActivities,
  TemporalModule,
} from "@btravstack/temporal-worker";
import { P } from "unthrown";

// The module forked around every activity attempt, SEEDED with the validated
// input on `ActivityInput(contract)`. The tenant the contract carries is
// claimed here once — not at each activity — and the application's vertical
// is composed over it, so `PlaceOrder` is bound before any activity runs.
const ActivityUnit = Module("ActivityUnit")({
  needs: [ActivityInput(contract)],
  imports: [OrderApplicationModule],
  provides: [
    Provider(Tenant)({
      inject: { input: ActivityInput(contract) },
      sync: ({ input }) => TenantId(input.tenantId),
    }),
  ],
  exports: [Tenant, PlaceOrder],
});

// The application's half: its activities, on the starter's own activities
// port, typed by the contract (a worker serves one activities record, so
// there is nothing to name). What the leaves read off `context.unit` is
// declared once, beside `inject`.
const orderActivities = TemporalActivities(contract)({
  unit: { place: PlaceOrder },
  sync: () => ({
    placeOrder: {
      place: ({ errors, context, input }) =>
        context.unit.place
          .execute(input.orderId, input.quantity)
          .mapErrCases((matcher) =>
            matcher
              .with(P.tag("DuplicateOrder"), (error) =>
                errors.OrderAlreadyPlaced({ id: error.id }),
              )
              .with(P.tag("InvalidOrderId"), (error) =>
                errors.InvalidOrderId({ id: error.id }),
              )
              .with(P.tag("InvalidQuantity"), (error) =>
                errors.InvalidQuantity({ id: error.id }),
              ),
          ),
    },
  }),
});

// The composition root: a di module, plus the contract, the activities
// provider and the workflow source — and nothing else to know.
const OrderWorker = TemporalModule("OrderWorker")({
  contract,
  activities: orderActivities,
  workflows: {
    workflowsPath: workflowsPathFromURL(import.meta.url, "./workflows.js"),
  },
  imports: [AppModule],
  unit: { activity: ActivityUnit },
});

await runMain(OrderWorker);

TemporalModule imports the starter — TemporalRuntime, TemporalConfig bound from TEMPORAL_ADDRESS (default 127.0.0.1:7233) and TEMPORAL_NAMESPACE (default default), TemporalConnection opened as a resource of the graph and closed on every exit path — provides the activities and exports the runtime port. A service that will not answer is a modeled TemporalUnreachable, exit 1; the starter calls declareActivitiesHandler itself, inside its own error qualifier, so a contract it cannot satisfy is a startup Err, not a defect. runtimeInfo() reads { taskQueue, namespace } back once the worker is polling.

Activity-piece DI identities include the contract task queue. Separate worker contracts can reuse keys without sharing a port; wire activity names stay unchanged.

A worker polling for several workflows can be several slices instead of one record: TemporalWorkflowActivities(contract, key)({ inject: { name: Dep }, unit, fork, sync }) mints a provider for ONE workflow's activities (or a contract-global activity), typed by the key alone, and TemporalActivities(contract)([...]) composes an array of them into the same activities provider TemporalModule takes — the array must cover every key the contract declares, and each piece's own port must still be discharged (provides), since the composed provider's deps are the pieces' ports, not what they close over. A workflow's activities belong to exactly one slice, and the compiler holds only half of that: coverage is checked, injectivity is not, so two slices claiming one key type-check together. It is di's duplicate-provider defect at build once both pieces are discharged as providers in one graph — wire in only one and the other's implementation is silently never registered.

A piece's unit names the ports its activities read off context.unit, resolved out of the per-attempt fork of the module bound on unit.activity — which is seeded with the validated input on ActivityInput(contract), so a unit module derives a tenant from the invocation rather than from an ambient record. The record arm above takes the same unit: — one record for every entry in it — so a worker that has not outgrown one function reaches context.unit without slicing first. The root is checked against what was declared either way: bind a module that does not export one of those ports and TemporalModule refuses it, naming the port.

A piece whose activities carry a different scope from the rest names its own module on fork: its attempts fork THAT module, seeded the same way, instead of unit.activity — a listing activity with no tenant in its input beside a fan-out whose activities each act for one. The piece's unit is then checked against its own module at the piece's call, the module's unmet needs become the root's (named by start), and an override reaches it under the module's name: overridden(root, providers, { unit: { TenantUnit: [...] } }).

Its own test suite needs a Docker daemon: the specs run against the shared temporalio/temporal development-server container internal/test-infra starts once per machine and every workspace reuses. Nothing else here does.

Diagnostic disclosure

withTemporalDiagnostics(options, work) scopes Temporal's diagnostic defaults without eagerly creating its native Runtime. Start the worker inside the callback and keep it open until owned workers and connections close. Startup refusals or defects before the first native connection then leave no diagnostic Runtime keeping the process alive. The helper returns an AsyncResult preserving the callback's outcome, restoring owned defaults on success, modeled error or defect, while preserving active Runtime instances and replacement configuration.

import { createLogger, diagnosticPolicy, jsonSink } from "@btravstack/observability";
import { withTemporalDiagnostics } from "@btravstack/temporal-worker";
import { fromSafePromise } from "unthrown";

await withTemporalDiagnostics(
  { logger: createLogger(jsonSink()), diagnostics: diagnosticPolicy() },
  () => fromSafePromise(runMain(OrderWorker)),
).get();

TemporalDiagnosticsOptions holds { logger, diagnostics, nativeLogLevel? }. SDK, workflow, activity and native Core logs reach the shared DiagnosticPolicy; native verbosity defaults to WARN, and the kernel owns process signals. An existing Runtime or overlapping scope is a defect. Do not call Runtime.install inside the callback: explicit SDK installation overrides scoped defaults. Temporal owns connection teardown and log flushing; the helper never closes an active Runtime or repairs native resources leaked after Runtime creation.

installTemporalDiagnostics(options) remains the eager, process-wide installer returning Result<void, never>. A late or repeated installation defects. It cannot release an idle Runtime after pre-connection startup failure; prefer the scoped helper for application startup.

temporalDiagnosticLogger(logger, diagnostics) returns a Temporal SDK logger for an application that configures Runtime.install itself. Native log forwarding still needs that Runtime configuration. Both helpers accept the policy declared in @btravstack/core; @btravstack/observability provides diagnosticPolicy(). See diagnostic disclosure for composition and the separate payload/failure boundary.

Schedules

@btravstack/temporal-worker/schedule exports ensureSchedule — the typed schedule client's create, made idempotent, so a deploy that runs again updates the schedule instead of failing on ScheduleAlreadyExistsError. @temporal-contract/client is an optional peer reached only through that subpath. Why scheduled work costs a cluster here, and when it should not be this stack at all: Run something on a schedule.

Testing

@btravstack/temporal-testing runs a worker's specs against one real Temporal server per Docker host, shared across test processes, with a namespace per spec file. It is a dev dependency of its own, so this package ships no test code. See Test a Temporal worker.

Options

• taskQueue: pins TEMPORAL_TASK_QUEUE, defaulting to the contract queue; see the reference.

TemporalModule(name)({...}) takes the starter's options plus activities and the module lists (imports, provides, exports, needs). It is the only way to compose the starter, which is not exported:

The full table — required/optional, defaults, and the reasoning — lives on the reference page, which is this list's one detailed home.

Securing the connection

TLS, mTLS and an API key are configured from the environment under Temporal's own names. For mTLS:

export TEMPORAL_TLS=true
export TEMPORAL_TLS_SERVER_CA_CERT_PATH=/var/run/secrets/temporal/ca.crt
export TEMPORAL_TLS_SERVER_NAME=temporal.example.internal
export TEMPORAL_TLS_CLIENT_CERT_PATH=/var/run/secrets/temporal/tls.crt
export TEMPORAL_TLS_CLIENT_KEY_PATH=/var/run/secrets/temporal/tls.key

or TEMPORAL_API_KEY_PATH (or TEMPORAL_API_KEY), which turns TLS on unless TEMPORAL_TLS says otherwise. Mounted files are read again every TEMPORAL_CREDENTIALS_RELOAD_MS (default 60000): a renewed certificate moves the worker to a new connection without a restart, a renewed key is set on the open one. A combination it cannot connect with is a ConfigInvalid at boot. The reference has the rules and the limits.

An application can supply renewable bearer credentials through an apiKey port implementing TemporalApiKeySource.read(). The source owns token acquisition, caching and expiry; the starter polls it with the credentials reload interval. A source requires TLS and excludes inline/file API keys and mTLS. See the reference for composition and fail-closed behavior.

The drain

worker.shutdown() stops polling at once, but run() resolves only when the last in-flight activity has finished — on Temporal's own shutdownForceTime. Serving.drain(signal) races that wait against the kernel's deadline and returns at whichever comes first: the kernel gets its thread back on time, the work is reported abandoned, and the worker keeps winding down on Temporal's clock until the process exits. @temporalio/worker exposes no public forced shutdown, so "stop waiting" is the escalation. Result → activity failure is declareActivitiesHandler's and is deliberately not duplicated here. The rest is on the documentation site.

TemporalModule automatically exports publicly available Logger and HealthChecks ports, including whole-module re-exports and providers declared at the root. It keeps private infrastructure hidden and business exports explicit.

License

MIT © Benoit TRAVERS