Skip to main content

worker

The temporal worker connects to the service and runs workflows and activities.

Usage​

import { NativeConnection, Worker } from '@temporalio/worker';
import * as activities from './activities';

async function run() {
  const connection = await NativeConnection.connect({
    address: 'localhost:7233',
  });
  try {
    const worker = await Worker.create({
      connection,
      namespace: 'default',
      taskQueue: 'sleep-for-days',
      workflowsPath: require.resolve('./workflows'),
      activities,
    });
    await worker.run();
  } finally {
    // Close the connection once the worker has stopped
    await connection.close();
  }
}

run().catch((err) => {
  console.error(err);
  process.exit(1);
});

Classes​

Interfaces​

Type Aliases​

ActivityInterceptorsFactory​

ActivityInterceptorsFactory = (ctx) => ActivityInterceptors

Instantiate Activity Interceptors for a given Temporal Activity execution.

This factory function is called for each Activity execution by this Worker, receiving the corresponding ActivityContext.

Note that there's no strict requirement to have different instances of Activity Interceptors for each activity execution. It is acceptable, and often sufficient, for this function to return the same instance of Activity Interceptors for all activity executions, ignoring the ctx parameter. Interceptor functions may then rely on the Context.current to access the Activity Context where needed.

Parameters​

ctx​

Context

Returns​

ActivityInterceptors

CompiledWorkerInterceptors​

CompiledWorkerInterceptors = Required<Pick<WorkerInterceptors, "client" | "activity" | "nexus" | "workflowModules">>

GetLogAttributesInput​

GetLogAttributesInput = Record<string, unknown>

Input for ActivityOutboundCallsInterceptor.getLogAttributes

GetMetricTagsInput​

GetMetricTagsInput = MetricTags

Input for ActivityOutboundCallsInterceptor.getMetricTags

GrpcCompressionConfig​

GrpcCompressionConfig = GzipGrpcCompressionConfig | NoneGrpcCompressionConfig

Transport-level gRPC compression configuration.

History​

History = IHistory

InjectedSink​

InjectedSink<T> = { [P in keyof T]: InjectedSinkFunction<T[P]> }

Converts a Sink from a mapping of name to function to a mapping of name to InjectedSinkFunction

Type Parameters​

T​

T extends Sink

InjectedSinks​

InjectedSinks<T> = { [P in keyof T]: InjectedSink<T[P]> }

Converts a Sinks interface from a mapping of name to Sink to a mapping of name to InjectedSink.

Used for type checking Sink injection against supplied type param T.

Type Parameters​

T​

T extends Sinks

LogExporterConfig​

LogExporterConfig = object & Partial<ConsoleLogger | ForwardLogger>

Configuration for logs emitted by the native side of the SDK.

Type Declaration​

filter?​

optional filter?: string | CoreLogFilterOptions

Determines the verboseness of log output emitted by the native side of the SDK.

This can be specified either as an (env filter format)[https://docs.rs/tracing-subscriber/0.2.20/tracing_subscriber/struct.EnvFilter.html] string, or as a CoreLogFilterOptions object.

Note that if log forwarding is enabled, then the configured Runtime.logger may apply further filtering on top of this.

BACKWARD COMPATIBILITY

If logging.filter is missing, the following legacy values (if present) will be used instead (in the given order):

  • ForwardLogger.forward.level => makeTelemetryFilterString({ core: level, other: level })
  • TelemetryOptions.tracingFilter
  • Default value of makeTelemetryFilterString({ core: 'WARN', other: 'ERROR'})
Default​

{ core: 'WARN', other: 'ERROR'} (with some exceptions, as described in backward compatibility note above).

See​

TelemetryOptions.logging

MetricsExporterConfig​

MetricsExporterConfig = object & PrometheusMetricsExporter | OtelCollectorExporter | BufferedMetricsExporter

Configuration for exporting metrics emitted by Core.

Type Declaration​

attachServiceName?​

optional attachServiceName?: boolean

Whether to put the service_name on every metric.

Default​
true
globalTags?​

optional globalTags?: Record<string, string>

Tags to add to all metrics emitted by the worker.

Note that this is not supported when the metrics are buffered.

metricPrefix?​

optional metricPrefix?: string

A prefix to add to all metrics.

Default​
'temporal_'
temporality?​

optional temporality?: "cumulative" | "delta"

Determines if the metrics exporter should use cumulative or delta temporality. Only applies to OpenTelemetry exporter.

Deprecated​

Use 'otel.temporality' instead

NexusInboundCallsInterceptor​

NexusInboundCallsInterceptor = object

Properties​

cancelOperation?​

optional cancelOperation?: (input, next) => Promise<void>

Parameters​
input​

NexusCancelOperationInput

next​

Next<NexusInboundCallsInterceptor, "cancelOperation">

Returns​

Promise<void>

startOperation?​

optional startOperation?: (input, next) => Promise<NexusStartOperationOutput>

Parameters​
input​

NexusStartOperationInput

next​

Next<NexusInboundCallsInterceptor, "startOperation">

Returns​

Promise<NexusStartOperationOutput>

NexusInterceptors​

NexusInterceptors = object

Interceptors for Nexus Operation execution.

Properties​

inbound?​

optional inbound?: NexusInboundCallsInterceptor

outbound?​

optional outbound?: NexusOutboundCallsInterceptor

NexusInterceptorsFactory​

NexusInterceptorsFactory = (ctx) => NexusInterceptors

Instantiate Nexus Interceptors for a given Nexus Operation execution.

This factory function is called for each Nexus Operation executed by this Worker, receiving the corresponding nexus.OperationContext | OperationContext.

Note that there's no strict requirement to have different instances of Nexus Interceptors for each Nexus Operation execution. It is acceptable and often sufficient for this function to return the same instance of Nexus Interceptors for all Nexus Operation executions, ignoring the ctx parameter. Interceptor functions may then rely on APIs provided by the @temporalio/nexus package to access the Nexus Operation Context where needed.

Parameters​

ctx​

nexus.OperationContext

Returns​

NexusInterceptors

NexusOutboundCallsInterceptor​

NexusOutboundCallsInterceptor = object

Properties​

getLogAttributes?​

optional getLogAttributes?: (input, next) => Record<string, unknown>

Parameters​
input​

GetLogAttributesInput

next​

Next<NexusOutboundCallsInterceptor, "getLogAttributes">

Returns​

Record<string, unknown>

getMetricTags?​

optional getMetricTags?: (input, next) => MetricTags

Parameters​
input​

GetMetricTagsInput

next​

Next<NexusOutboundCallsInterceptor, "getMetricTags">

Returns​

MetricTags

PatchActivationCallback​

PatchActivationCallback = (input) => boolean

Experimental

Parameters​

input​

PatchActivationInput

Returns​

boolean

ProxyConfig​

ProxyConfig = HttpConnectProxyConfig

ReplayHistoriesIterable​

ReplayHistoriesIterable = AsyncIterable<HistoryAndWorkflowId> | Iterable<HistoryAndWorkflowId>

An iterable on workflow histories and their IDs, used for batch replaying.

ResourceBasedSlotsForType​

ResourceBasedSlotsForType = ResourceBasedSlotOptions & ResourceBasedTunerConfig & object

Resource based slot supplier options for a specific kind of slot.

Type Declaration​

type​

type: "resource-based"

SlotInfo​

SlotInfo = WorkflowSlotInfo | ActivitySlotInfo | LocalActivitySlotInfo | NexusSlotInfo

SlotSupplier​

SlotSupplier<SI> = ResourceBasedSlotsForType | FixedSizeSlotSupplier | CustomSlotSupplier<SI>

Controls how slots are handed out for a specific task type.

Type Parameters​

SI​

SI extends SlotInfo

State​

State = "INITIALIZED" | "RUNNING" | "STOPPED" | "STOPPING" | "DRAINING" | "DRAINED" | "FAILED"

The worker's possible states

  • INITIALIZED - The initial state of the Worker after calling Worker.create and successful connection to the server
  • RUNNING - Worker.run was called, polling task queues
  • STOPPING - Worker.shutdown was called or received shutdown signal, worker will forcefully shutdown in shutdownGraceTime
  • DRAINING - Core has indicated that shutdown is complete and all Workflow tasks have been drained, waiting for activities and cached workflows eviction
  • DRAINED - All activities and workflows have completed, ready to shutdown
  • STOPPED - Shutdown complete, Worker.run resolves
  • FAILED - Worker encountered an unrecoverable error, Worker.run should reject with the error

WorkerDeploymentOptions​

WorkerDeploymentOptions = object & { defaultVersioningBehavior: VersioningBehavior; useWorkerVersioning: true; } | { defaultVersioningBehavior?: never; useWorkerVersioning: false; }

Allows specifying the deployment version of the worker and whether to use deployment-based worker versioning.

Type Declaration​

defaultVersioningBehavior?​

optional defaultVersioningBehavior?: VersioningBehavior

The default versioning behavior to use for all workflows on this worker.

Required if useWorkerVersioning is true; should be left unset otherwise.

useWorkerVersioning​

useWorkerVersioning: boolean

Whether to use deployment-based worker versioning.

version​

version: WorkerDeploymentVersion

The deployment version of the worker.

WorkerTuner​

WorkerTuner = ResourceBasedTuner | ResourceBasedTunerWithController | TunerHolder

A worker tuner allows the customization of the performance characteristics of workers by controlling how "slots" are handed out for different task types. In order to poll for and then run tasks, a slot must first be reserved by the SlotSupplier returned by the tuner.

WorkflowBundleOption​

WorkflowBundleOption = WorkflowBundle | WorkflowBundleWithSourceMap | WorkflowBundlePath | WorkflowBundlePathWithSourceMap

Variables​

defaultWorkflowInterceptorModules​

const defaultWorkflowInterceptorModules: never[]

Deprecated​

Including defaultWorkflowInterceptorModules in BundlerOptions.workflowInterceptorModules is no longer required.

errors​

const errors: object

Type Declaration​

GracefulShutdownPeriodExpiredError​

GracefulShutdownPeriodExpiredError: typeof GracefulShutdownPeriodExpiredError

IllegalStateError​

IllegalStateError: typeof IllegalStateError

ShutdownError​

ShutdownError: typeof bridgeErrors.ShutdownError

TransportError​

TransportError: typeof bridgeErrors.TransportError

UnexpectedError​

UnexpectedError: typeof bridgeErrors.UnexpectedError

Deprecated​

Import error classes directly

LogTimestamp​

const LogTimestamp: unique symbol

ShutdownError​

const ShutdownError: typeof bridgeErrors.ShutdownError

TransportError​

const TransportError: typeof bridgeErrors.TransportError

UnexpectedError​

const UnexpectedError: typeof bridgeErrors.UnexpectedError

WorkflowInboundLogInterceptor​

const WorkflowInboundLogInterceptor: typeof WorkflowLogInterceptor

Deprecated​

WorkflowInboundLogInterceptor is deprecated. Workflow lifecycle events are now automatically logged by the SDK. To customize workflow log attributes, simply register a custom WorkflowInterceptors that intercepts the outbound.getLogAttributes() method.

Functions​

activityLogAttributes()​

activityLogAttributes(info): Record<string, unknown>

Returns a map of attributes to be set on log messages for a given Activity

Parameters​

info​

Info

Returns​

Record<string, unknown>

appendDefaultInterceptors()​

appendDefaultInterceptors(interceptors, logger?): WorkerInterceptors

Appends the default Worker logging interceptors to given interceptor arrays.

Parameters​

interceptors​

WorkerInterceptors

logger?​

Logger

a Logger - defaults to the Runtime singleton logger.

Returns​

WorkerInterceptors

Deprecated​

Calling appendDefaultInterceptors() is no longer required. To configure a custom logger, set the Runtime.logger property instead.

bundleWorkflowCode()​

bundleWorkflowCode(options): Promise<WorkflowBundleWithSourceMap>

Create a bundle to pass to WorkerOptions.workflowBundle. Helpful for reducing Worker startup time in production.

When using with Worker.runReplayHistory, make sure to pass the same interceptors and payload converter used when the history was generated.

Parameters​

options​

BundleOptions

Returns​

Promise<WorkflowBundleWithSourceMap>

defaultSinks()​

defaultSinks(logger?): InjectedSinks<LoggerSinks>

Build the sink used internally by the SDK to forwards log messages from the Workflow sandbox to an actual logger.

Parameters​

logger?​

Logger

a Logger - defaults to the Runtime singleton logger.

Returns​

InjectedSinks<LoggerSinks>

Deprecated​

Calling defaultSink() is no longer required. To configure a custom logger, set the Runtime.logger property instead.

makeTelemetryFilterString()​

makeTelemetryFilterString(options): string

A helper to build a filter string for use in RuntimeOptions.telemetryOptions.tracingFilter.

Note that one may instead simply pass a CoreLogFilterOptions object directly to RuntimeOptions.telemetryOptions.logging.filter. This function may however still be useful in some particular use cases and will therefore be kept around.

Parameters​

options​

CoreLogFilterOptions

Returns​

string

startDebugReplayer()​

startDebugReplayer(options): void

Start a replayer for debugging purposes.

Use this method to integrate the replayer with external debuggers like the Temporal VS Code debbuger extension.

Parameters​

options​

ReplayWorkerOptions

Returns​

void

workflowLogAttributes()​

workflowLogAttributes(info): Record<string, unknown>

Returns a map of attributes to be set by default on log messages for a given Workflow. Note that this function may be called from outside of the Workflow context (eg. by the worker itself).

Parameters​

info​

WorkflowInfo

Returns​

Record<string, unknown>

References​

DataConverter​

Re-exports DataConverter

defaultPayloadConverter​

Re-exports defaultPayloadConverter

Headers​

Re-exports Headers

IllegalStateError​

Re-exports IllegalStateError

Logger​

Re-exports Logger

LoggerSinks​

Re-exports LoggerSinks

LogLevel​

Re-exports LogLevel

LogMetadata​

Re-exports LogMetadata

MetricsExporter​

Renames and re-exports MetricsExporterConfig

Next​

Re-exports Next

PayloadCache​

Re-exports PayloadCache