Socket
Socket
Sign inDemoInstall

@temporalio/workflow

Package Overview
Dependencies
Maintainers
0
Versions
79
Alerts
File Explorer

Advanced tools

License
Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

@temporalio/workflow - npm Package Compare versions

Comparing version 1.10.3 to 1.11.0

lib/update-scope.d.ts
lib/update-scope.js
lib/update-scope.js.map
src/update-scope.ts

18

lib/flags.d.ts

@@ -17,3 +17,21 @@ export type SdkFlag = {

readonly NonCancellableScopesAreShieldedFromPropagation: SdkFlag;
/**
* Prior to 1.11.0, when processing a Workflow activation, the SDK would execute `notifyHasPatch`
* and `signalWorkflow` jobs in distinct phases, before other types of jobs. The primary reason
* behind that multi-phase algorithm was to avoid the possibility that a Workflow execution might
* complete before all incoming signals have been dispatched (at least to the point that the
* _synchronous_ part of the handler function has been executed).
*
* This flag replaces that multi-phase algorithm with a simpler one where jobs are simply sorted as
* `(signals and updates) -> others`, but without processing them as distinct batches (i.e. without
* leaving/reentering the VM context between each group, which automatically triggers the execution
* of all outstanding microtasks). That single-phase approach resolves a number of quirks of the
* former algorithm, and yet still satisfies to the original requirement of ensuring that every
* `signalWorkflow` jobs - and now `doUpdate` jobs as well - have been given a proper chance to
* execute before the Workflow main function might completes.
*
* @since Introduced in 1.11.0. This change is not rollback-safe.
*/
readonly ProcessWorkflowActivationJobsAsSingleBatch: SdkFlag;
};
export declare function assertValidFlag(id: number): void;

20

lib/flags.js

@@ -16,3 +16,21 @@ "use strict";

*/
NonCancellableScopesAreShieldedFromPropagation: defineFlag(1, false),
NonCancellableScopesAreShieldedFromPropagation: defineFlag(1, true),
/**
* Prior to 1.11.0, when processing a Workflow activation, the SDK would execute `notifyHasPatch`
* and `signalWorkflow` jobs in distinct phases, before other types of jobs. The primary reason
* behind that multi-phase algorithm was to avoid the possibility that a Workflow execution might
* complete before all incoming signals have been dispatched (at least to the point that the
* _synchronous_ part of the handler function has been executed).
*
* This flag replaces that multi-phase algorithm with a simpler one where jobs are simply sorted as
* `(signals and updates) -> others`, but without processing them as distinct batches (i.e. without
* leaving/reentering the VM context between each group, which automatically triggers the execution
* of all outstanding microtasks). That single-phase approach resolves a number of quirks of the
* former algorithm, and yet still satisfies to the original requirement of ensuring that every
* `signalWorkflow` jobs - and now `doUpdate` jobs as well - have been given a proper chance to
* execute before the Workflow main function might completes.
*
* @since Introduced in 1.11.0. This change is not rollback-safe.
*/
ProcessWorkflowActivationJobsAsSingleBatch: defineFlag(2, true),
};

@@ -19,0 +37,0 @@ function defineFlag(id, def) {

2

lib/index.d.ts

@@ -60,3 +60,3 @@ /**

export * from './interceptors';
export { ChildWorkflowCancellationType, ChildWorkflowOptions, ContinueAsNew, ContinueAsNewOptions, EnhancedStackTrace, FileLocation, FileSlice, ParentClosePolicy, ParentWorkflowInfo, SDKInfo, StackTrace, UnsafeWorkflowInfo, WorkflowInfo, } from './interfaces';
export { ChildWorkflowCancellationType, ChildWorkflowOptions, ContinueAsNew, ContinueAsNewOptions, EnhancedStackTrace, StackTraceFileLocation, StackTraceFileSlice, ParentClosePolicy, ParentWorkflowInfo, StackTraceSDKInfo, StackTrace, UnsafeWorkflowInfo, WorkflowInfo, } from './interfaces';
export { proxySinks, Sink, SinkCall, SinkFunction, Sinks } from './sinks';

@@ -63,0 +63,0 @@ export { log } from './logs';

48

lib/interfaces.d.ts
import type { RawSourceMap } from 'source-map';
import { RetryPolicy, TemporalFailure, CommonWorkflowOptions, SearchAttributes, SignalDefinition, UpdateDefinition, QueryDefinition, Duration, VersioningIntent } from '@temporalio/common';
import { RetryPolicy, TemporalFailure, CommonWorkflowOptions, HandlerUnfinishedPolicy, SearchAttributes, SignalDefinition, UpdateDefinition, QueryDefinition, Duration, VersioningIntent } from '@temporalio/common';
import type { coresdk } from '@temporalio/proto';

@@ -152,2 +152,17 @@ /**

}
/**
* Information about a workflow update.
*
* @experimental
*/
export interface UpdateInfo {
/**
* A workflow-unique identifier for this update.
*/
readonly id: string;
/**
* The update type name.
*/
readonly name: string;
}
export interface ParentWorkflowInfo {

@@ -302,3 +317,3 @@ workflowId: string;

export type ChildWorkflowOptionsWithDefaults = ChildWorkflowOptions & RequiredChildWorkflowOptions;
export interface SDKInfo {
export interface StackTraceSDKInfo {
name: string;

@@ -310,11 +325,11 @@ version: string;

*/
export interface FileSlice {
export interface StackTraceFileSlice {
/**
* Only used possible to trim the file without breaking syntax highlighting.
*/
line_offset: number;
/**
* slice of a file with `\n` (newline) line terminator.
*/
content: string;
/**
* Only used possible to trim the file without breaking syntax highlighting.
*/
lineOffset: number;
}

@@ -324,3 +339,3 @@ /**

*/
export interface FileLocation {
export interface StackTraceFileLocation {
/**

@@ -330,3 +345,3 @@ * Path to source file (absolute or relative).

*/
filePath?: string;
file_path?: string;
/**

@@ -344,6 +359,10 @@ * If possible, SDK should send this, required for displaying the code location.

*/
functionName?: string;
function_name?: string;
/**
* Flag to mark this as internal SDK code and hide by default in the UI.
*/
internal_code: boolean;
}
export interface StackTrace {
locations: FileLocation[];
locations: StackTraceFileLocation[];
}

@@ -354,3 +373,3 @@ /**

export interface EnhancedStackTrace {
sdk: SDKInfo;
sdk: StackTraceSDKInfo;
/**

@@ -361,3 +380,3 @@ * Mapping of file path to file contents.

*/
sources: Record<string, FileSlice[]>;
sources: Record<string, StackTraceFileSlice[]>;
stacks: StackTrace[];

@@ -370,2 +389,3 @@ }

patches: string[];
sdkFlags: number[];
showStackTraceSources: boolean;

@@ -401,2 +421,3 @@ }

description?: string;
unfinishedPolicy?: HandlerUnfinishedPolicy;
};

@@ -409,2 +430,3 @@ /**

description?: string;
unfinishedPolicy?: HandlerUnfinishedPolicy;
};

@@ -411,0 +433,0 @@ export interface ActivationCompletion {

78

lib/internals.d.ts
import type { RawSourceMap } from 'source-map';
import { FailureConverter, PayloadConverter, Workflow, WorkflowQueryAnnotatedType, WorkflowSignalAnnotatedType, WorkflowUpdateAnnotatedType, ProtoFailure } from '@temporalio/common';
import { FailureConverter, PayloadConverter, HandlerUnfinishedPolicy, Workflow, WorkflowQueryAnnotatedType, WorkflowSignalAnnotatedType, WorkflowUpdateAnnotatedType, ProtoFailure, WorkflowUpdateType, WorkflowUpdateValidatorType } from '@temporalio/common';
import type { coresdk } from '@temporalio/proto';

@@ -7,3 +7,3 @@ import { RNG } from './alea';

import { QueryInput, SignalInput, UpdateInput, WorkflowExecuteInput, WorkflowInterceptors } from './interceptors';
import { DefaultSignalHandler, FileLocation, WorkflowInfo, WorkflowCreateOptionsInternal, ActivationCompletion } from './interfaces';
import { DefaultSignalHandler, StackTraceFileLocation, WorkflowInfo, WorkflowCreateOptionsInternal, ActivationCompletion } from './interfaces';
import { type SinkCall } from './sinks';

@@ -13,3 +13,3 @@ import { SdkFlag } from './flags';

formatted: string;
structured: FileLocation[];
structured: StackTraceFileLocation[];
}

@@ -39,5 +39,26 @@ /**

/**
* Information about an update or signal handler execution.
*/
interface MessageHandlerExecution {
name: string;
unfinishedPolicy: HandlerUnfinishedPolicy;
id?: string;
}
/**
* Keeps all of the Workflow runtime state like pending completions for activities and timers.
*
* Implements handlers for all workflow activation jobs.
*
* Note that most methods in this class are meant to be called only from within the VM.
*
* However, a few methods may be called directly from outside the VM (essentially from `vm-shared.ts`).
* These methods are specifically marked with a comment and require careful consideration, as the
* execution context may not properly reflect that of the target workflow execution (e.g.: with Reusable
* VMs, the `global` may not have been swapped to those of that workflow execution; the active microtask
* queue may be that of the thread/process, rather than the queue of that VM context; etc). Consequently,
* methods that are meant to be called from outside of the VM must not do any of the following:
*
* - Access any global variable;
* - Create Promise objects, use async/await, or otherwise schedule microtasks;
* - Call user-defined functions, including any form of interceptor.
*/

@@ -69,10 +90,2 @@ export declare class Activator implements ActivationHandler {

/**
* Holds buffered query calls until a handler is registered.
*
* **IMPORTANT** queries are only buffered until workflow is started.
* This is required because async interceptors might block workflow function invocation
* which delays query handler registration.
*/
protected readonly bufferedQueries: coresdk.workflow_activation.IQueryWorkflow[];
/**
* Mapping of update name to handler and validator

@@ -86,2 +99,14 @@ */

/**
* Mapping of in-progress updates to handler execution information.
*/
readonly inProgressUpdates: Map<string, MessageHandlerExecution>;
/**
* Mapping of in-progress signals to handler execution information.
*/
readonly inProgressSignals: Map<number, MessageHandlerExecution>;
/**
* A sequence number providing unique identifiers for signal handler executions.
*/
protected signalHandlerExecutionSeq: number;
/**
* A signal handler that catches calls for non-registered signal names.

@@ -129,8 +154,2 @@ */

/**
* This is tracked to allow buffering queries until a workflow function is called.
* TODO(bergundy): I don't think this makes sense since queries run last in an activation and must be responded to in
* the same activation.
*/
protected workflowFunctionWasCalled: boolean;
/**
* The next (incremental) sequence to assign when generating completable commands

@@ -149,2 +168,3 @@ */

* This is set every time the workflow executes an activation
* May be accessed and modified from outside the VM.
*/

@@ -158,2 +178,3 @@ now: number;

* Information about the current Workflow
* May be accessed from outside the VM.
*/

@@ -185,5 +206,11 @@ info: WorkflowInfo;

readonly registeredActivityNames: Set<string>;
constructor({ info, now, showStackTraceSources, sourceMap, getTimeOfDay, randomnessSeed, patches, registeredActivityNames, }: WorkflowCreateOptionsInternal);
constructor({ info, now, showStackTraceSources, sourceMap, getTimeOfDay, sdkFlags, randomnessSeed, patches, registeredActivityNames, }: WorkflowCreateOptionsInternal);
/**
* May be invoked from outside the VM.
*/
mutateWorkflowInfo(fn: (info: WorkflowInfo) => WorkflowInfo): void;
protected getStackTraces(): Stack[];
/**
* May be invoked from outside the VM.
*/
getAndResetSinkCalls(): SinkCall[];

@@ -207,4 +234,4 @@ /**

doUpdate(activation: coresdk.workflow_activation.IDoUpdate): void;
protected updateNextHandler({ name, args }: UpdateInput): Promise<unknown>;
protected validateUpdateNextHandler({ name, args }: UpdateInput): void;
protected updateNextHandler(handler: WorkflowUpdateType, { args }: UpdateInput): Promise<unknown>;
protected validateUpdateNextHandler(validator: WorkflowUpdateValidatorType | undefined, { args }: UpdateInput): void;
dispatchBufferedUpdates(): void;

@@ -217,6 +244,16 @@ rejectBufferedUpdates(): void;

resolveRequestCancelExternalWorkflow(activation: coresdk.workflow_activation.IResolveRequestCancelExternalWorkflow): void;
warnIfUnfinishedHandlers(): void;
updateRandomSeed(activation: coresdk.workflow_activation.IUpdateRandomSeed): void;
notifyHasPatch(activation: coresdk.workflow_activation.INotifyHasPatch): void;
patchInternal(patchId: string, deprecated: boolean): boolean;
/**
* Called early while handling an activation to register known flags.
* May be invoked from outside the VM.
*/
addKnownFlags(flags: number[]): void;
/**
* Check if a flag is known to the Workflow Execution; if not, enable the flag if workflow
* is not replaying and the flag is configured to be enabled by default.
* May be invoked from outside the VM.
*/
hasFlag(flag: SdkFlag): boolean;

@@ -242,1 +279,2 @@ removeFromCache(): void;

}
export {};

234

lib/internals.js

@@ -12,2 +12,3 @@ "use strict";

const cancellation_scope_1 = require("./cancellation-scope");
const update_scope_1 = require("./update-scope");
const errors_1 = require("./errors");

@@ -17,4 +18,4 @@ const interfaces_1 = require("./interfaces");

const pkg_1 = __importDefault(require("./pkg"));
const flags_1 = require("./flags");
const logs_1 = require("./logs");
const flags_1 = require("./flags");
var StartChildWorkflowExecutionFailedCause;

@@ -31,5 +32,18 @@ (function (StartChildWorkflowExecutionFailedCause) {

* Implements handlers for all workflow activation jobs.
*
* Note that most methods in this class are meant to be called only from within the VM.
*
* However, a few methods may be called directly from outside the VM (essentially from `vm-shared.ts`).
* These methods are specifically marked with a comment and require careful consideration, as the
* execution context may not properly reflect that of the target workflow execution (e.g.: with Reusable
* VMs, the `global` may not have been swapped to those of that workflow execution; the active microtask
* queue may be that of the thread/process, rather than the queue of that VM context; etc). Consequently,
* methods that are meant to be called from outside of the VM must not do any of the following:
*
* - Access any global variable;
* - Create Promise objects, use async/await, or otherwise schedule microtasks;
* - Call user-defined functions, including any form of interceptor.
*/
class Activator {
constructor({ info, now, showStackTraceSources, sourceMap, getTimeOfDay, randomnessSeed, patches, registeredActivityNames, }) {
constructor({ info, now, showStackTraceSources, sourceMap, getTimeOfDay, sdkFlags, randomnessSeed, patches, registeredActivityNames, }) {
/**

@@ -59,10 +73,2 @@ * Cache for modules - referenced in reusable-vm.ts

/**
* Holds buffered query calls until a handler is registered.
*
* **IMPORTANT** queries are only buffered until workflow is started.
* This is required because async interceptors might block workflow function invocation
* which delays query handler registration.
*/
this.bufferedQueries = Array();
/**
* Mapping of update name to handler and validator

@@ -75,2 +81,14 @@ */

this.signalHandlers = new Map();
/**
* Mapping of in-progress updates to handler execution information.
*/
this.inProgressUpdates = new Map();
/**
* Mapping of in-progress signals to handler execution information.
*/
this.inProgressSignals = new Map();
/**
* A sequence number providing unique identifiers for signal handler executions.
*/
this.signalHandlerExecutionSeq = 0;
this.promiseStackStore = {

@@ -106,12 +124,12 @@ promiseToStack: new Map(),

for (const { locations } of stacks) {
for (const { filePath } of locations) {
if (!filePath)
for (const { file_path } of locations) {
if (!file_path)
continue;
const content = sourceMap?.sourcesContent?.[sourceMap?.sources.indexOf(filePath)];
const content = sourceMap?.sourcesContent?.[sourceMap?.sources.indexOf(file_path)];
if (!content)
continue;
sources[filePath] = [
sources[file_path] = [
{
line_offset: 0,
content,
lineOffset: 0,
},

@@ -147,3 +165,2 @@ ];

type: workflowType,
description: null, // For now, do not set the workflow description in the TS SDK.
queryDefinitions,

@@ -162,3 +179,7 @@ signalDefinitions,

*/
this.interceptors = { inbound: [], outbound: [], internals: [] };
this.interceptors = {
inbound: [],
outbound: [],
internals: [],
};
/**

@@ -185,8 +206,2 @@ * Buffer that stores all generated commands, reset after each activation

/**
* This is tracked to allow buffering queries until a workflow function is called.
* TODO(bergundy): I don't think this makes sense since queries run last in an activation and must be responded to in
* the same activation.
*/
this.workflowFunctionWasCalled = false;
/**
* The next (incremental) sequence to assign when generating completable commands

@@ -226,8 +241,10 @@ */

this.registeredActivityNames = registeredActivityNames;
if (info.unsafe.isReplaying) {
for (const patchId of patches) {
this.notifyHasPatch({ patchId });
}
this.addKnownFlags(sdkFlags);
for (const patchId of patches) {
this.notifyHasPatch({ patchId });
}
}
/**
* May be invoked from outside the VM.
*/
mutateWorkflowInfo(fn) {

@@ -258,2 +275,5 @@ this.info = fn(this.info);

}
/**
* May be invoked from outside the VM.
*/
getAndResetSinkCalls() {

@@ -270,5 +290,2 @@ const { sinkCalls } = this;

pushCommand(cmd, complete = false) {
// Only query responses may be sent after completion
if (this.completed && !cmd.respondToQuery)
return;
this.commands.push(cmd);

@@ -290,16 +307,3 @@ if (complete) {

}
let promise;
try {
promise = workflow(...args);
}
finally {
// Queries must be handled even if there was an exception when invoking the Workflow function.
this.workflowFunctionWasCalled = true;
// Empty the buffer
const buffer = this.bufferedQueries.splice(0);
for (const activation of buffer) {
this.queryWorkflow(activation);
}
}
return await promise;
return await workflow(...args);
}

@@ -417,6 +421,2 @@ startWorkflow(activation) {

queryWorkflow(activation) {
if (!this.workflowFunctionWasCalled) {
this.bufferedQueries.push(activation);
return;
}
const { queryType, queryId, headers } = activation;

@@ -445,3 +445,4 @@ if (!(queryType && queryId)) {

}
if (!this.updateHandlers.has(name)) {
const entry = this.updateHandlers.get(name);
if (!entry) {
this.bufferedUpdates.push(activation);

@@ -483,37 +484,39 @@ return;

// These are caught elsewhere and fail the corresponding activation.
let input;
try {
if (runValidator && this.updateHandlers.get(name)?.validator) {
const validate = (0, interceptors_1.composeInterceptors)(this.interceptors.inbound, 'validateUpdate', this.validateUpdateNextHandler.bind(this));
validate(makeInput());
const doUpdateImpl = async () => {
let input;
try {
if (runValidator && entry.validator) {
const validate = (0, interceptors_1.composeInterceptors)(this.interceptors.inbound, 'validateUpdate', this.validateUpdateNextHandler.bind(this, entry.validator));
validate(makeInput());
}
input = makeInput();
}
input = makeInput();
}
catch (error) {
this.rejectUpdate(protocolInstanceId, error);
return;
}
const execute = (0, interceptors_1.composeInterceptors)(this.interceptors.inbound, 'handleUpdate', this.updateNextHandler.bind(this));
this.acceptUpdate(protocolInstanceId);
(0, stack_helpers_1.untrackPromise)(execute(input)
.then((result) => this.completeUpdate(protocolInstanceId, result))
.catch((error) => {
if (error instanceof common_1.TemporalFailure) {
catch (error) {
this.rejectUpdate(protocolInstanceId, error);
return;
}
else {
throw error;
}
}));
this.acceptUpdate(protocolInstanceId);
const execute = (0, interceptors_1.composeInterceptors)(this.interceptors.inbound, 'handleUpdate', this.updateNextHandler.bind(this, entry.handler));
const { unfinishedPolicy } = entry;
this.inProgressUpdates.set(updateId, { name, unfinishedPolicy, id: updateId });
const res = execute(input)
.then((result) => this.completeUpdate(protocolInstanceId, result))
.catch((error) => {
if (error instanceof common_1.TemporalFailure) {
this.rejectUpdate(protocolInstanceId, error);
}
else {
throw error;
}
})
.finally(() => this.inProgressUpdates.delete(updateId));
(0, stack_helpers_1.untrackPromise)(res);
return res;
};
(0, stack_helpers_1.untrackPromise)(update_scope_1.UpdateScope.updateWithInfo(updateId, name, doUpdateImpl));
}
async updateNextHandler({ name, args }) {
const entry = this.updateHandlers.get(name);
if (!entry) {
return Promise.reject(new common_1.IllegalStateError(`No registered update handler for update: ${name}`));
}
const { handler } = entry;
async updateNextHandler(handler, { args }) {
return await handler(...args);
}
validateUpdateNextHandler({ name, args }) {
const { validator } = this.updateHandlers.get(name) ?? {};
validateUpdateNextHandler(validator, { args }) {
if (validator) {

@@ -566,2 +569,8 @@ validator(...args);

}
// If we fall through to the default signal handler then the unfinished
// policy is WARN_AND_ABANDON; users currently have no way to silence any
// ensuing warnings.
const unfinishedPolicy = this.signalHandlers.get(signalName)?.unfinishedPolicy ?? common_1.HandlerUnfinishedPolicy.WARN_AND_ABANDON;
const signalExecutionNum = this.signalHandlerExecutionSeq++;
this.inProgressSignals.set(signalExecutionNum, { name: signalName, unfinishedPolicy });
const execute = (0, interceptors_1.composeInterceptors)(this.interceptors.inbound, 'handleSignal', this.signalWorkflowNextHandler.bind(this));

@@ -572,3 +581,5 @@ execute({

headers: headers ?? {},
}).catch(this.handleWorkflowFailure.bind(this));
})
.catch(this.handleWorkflowFailure.bind(this))
.finally(() => this.inProgressSignals.delete(signalExecutionNum));
}

@@ -610,2 +621,15 @@ dispatchBufferedSignals() {

}
warnIfUnfinishedHandlers() {
const getWarnable = (handlerExecutions) => {
return Array.from(handlerExecutions).filter((ex) => ex.unfinishedPolicy === common_1.HandlerUnfinishedPolicy.WARN_AND_ABANDON);
};
const warnableUpdates = getWarnable(this.inProgressUpdates.values());
if (warnableUpdates.length > 0) {
logs_1.log.warn(makeUnfinishedUpdateHandlerMessage(warnableUpdates));
}
const warnableSignals = getWarnable(this.inProgressSignals.values());
if (warnableSignals.length > 0) {
logs_1.log.warn(makeUnfinishedSignalHandlerMessage(warnableSignals));
}
}
updateRandomSeed(activation) {

@@ -618,5 +642,6 @@ if (!activation.randomnessSeed) {

notifyHasPatch(activation) {
if (!activation.patchId) {
throw new TypeError('Notify has patch missing patch name');
}
if (!this.info.unsafe.isReplaying)
throw new common_1.IllegalStateError('Unexpected notifyHasPatch job on non-replay activation');
if (!activation.patchId)
throw new TypeError('notifyHasPatch missing patch id');
this.knownPresentPatches.add(activation.patchId);

@@ -639,3 +664,6 @@ }

}
// Called early while handling an activation to register known flags
/**
* Called early while handling an activation to register known flags.
* May be invoked from outside the VM.
*/
addKnownFlags(flags) {

@@ -647,2 +675,7 @@ for (const flag of flags) {

}
/**
* Check if a flag is known to the Workflow Execution; if not, enable the flag if workflow
* is not replaying and the flag is configured to be enabled by default.
* May be invoked from outside the VM.
*/
hasFlag(flag) {

@@ -678,2 +711,6 @@ if (this.knownFlags.has(flag.id)) {

}
// Fail the workflow. We do not want to issue unfinishedHandlers warnings. To achieve that, we
// mark all handlers as completed now.
this.inProgressSignals.clear();
this.inProgressUpdates.clear();
this.pushCommand({

@@ -753,2 +790,33 @@ failWorkflowExecution: {

}
function makeUnfinishedUpdateHandlerMessage(handlerExecutions) {
const message = `
[TMPRL1102] Workflow finished while an update handler was still running. This may have interrupted work that the
update handler was doing, and the client that sent the update will receive a 'workflow execution
already completed' RPCError instead of the update result. You can wait for all update and signal
handlers to complete by using \`await workflow.condition(workflow.allHandlersFinished)\`.
Alternatively, if both you and the clients sending the update are okay with interrupting running handlers
when the workflow finishes, and causing clients to receive errors, then you can disable this warning by
passing an option when setting the handler:
\`workflow.setHandler(myUpdate, myUpdateHandler, {unfinishedPolicy: HandlerUnfinishedPolicy.ABANDON});\`.`
.replace(/\n/g, ' ')
.trim();
return `${message} The following updates were unfinished (and warnings were not disabled for their handler): ${JSON.stringify(handlerExecutions.map((ex) => ({ name: ex.name, id: ex.id })))}`;
}
function makeUnfinishedSignalHandlerMessage(handlerExecutions) {
const message = `
[TMPRL1102] Workflow finished while a signal handler was still running. This may have interrupted work that the
signal handler was doing. You can wait for all update and signal handlers to complete by using
\`await workflow.condition(workflow.allHandlersFinished)\`. Alternatively, if both you and the
clients sending the update are okay with interrupting running handlers when the workflow finishes,
then you can disable this warning by passing an option when setting the handler:
\`workflow.setHandler(mySignal, mySignalHandler, {unfinishedPolicy: HandlerUnfinishedPolicy.ABANDON});\`.`
.replace(/\n/g, ' ')
.trim();
const names = new Map();
for (const ex of handlerExecutions) {
const count = names.get(ex.name) || 0;
names.set(ex.name, count + 1);
}
return `${message} The following signals were unfinished (and warnings were not disabled for their handler): ${JSON.stringify(Array.from(names.entries()).map(([name, count]) => ({ name, count })))}`;
}
//# sourceMappingURL=internals.js.map

9

lib/worker-interface.d.ts
import { coresdk } from '@temporalio/proto';
import { WorkflowCreateOptionsInternal } from './interfaces';
import { type SinkCall } from './sinks';
export { PromiseStackStore } from './internals';

@@ -13,5 +12,4 @@ /**

* Run a chunk of activation jobs
* @returns a boolean indicating whether job was processed or ignored
*/
export declare function activate(activation: coresdk.workflow_activation.WorkflowActivation, batchIndex: number): void;
export declare function activate(activation: coresdk.workflow_activation.IWorkflowActivation, batchIndex: number): void;
/**

@@ -24,3 +22,2 @@ * Conclude a single activation.

export declare function concludeActivation(): coresdk.workflow_completion.IWorkflowActivationCompletion;
export declare function getAndResetSinkCalls(): SinkCall[];
/**

@@ -32,6 +29,2 @@ * Loop through all blocked conditions, evaluate and unblock if possible.

export declare function tryUnblockConditions(): number;
/**
* Predicate used to prevent triggering conditions for non-query and non-patch jobs.
*/
export declare function shouldUnblockConditions(job: coresdk.workflow_activation.IWorkflowActivationJob): boolean;
export declare function dispose(): void;

71

lib/worker-interface.js
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.dispose = exports.shouldUnblockConditions = exports.tryUnblockConditions = exports.getAndResetSinkCalls = exports.concludeActivation = exports.activate = exports.initRuntime = void 0;
exports.dispose = exports.tryUnblockConditions = exports.concludeActivation = exports.activate = exports.initRuntime = void 0;
/**

@@ -10,5 +10,5 @@ * Exported functions for the Worker to interact with the Workflow isolate

const common_1 = require("@temporalio/common");
const time_1 = require("@temporalio/common/lib/time");
const interceptors_1 = require("@temporalio/common/lib/interceptors");
const cancellation_scope_1 = require("./cancellation-scope");
const update_scope_1 = require("./update-scope");
const internals_1 = require("./internals");

@@ -31,3 +31,3 @@ const global_attributes_1 = require("./global-attributes");

});
// There's on activator per workflow instance, set it globally on the context.
// There's one activator per workflow instance, set it globally on the context.
// We do this before importing any user code so user code can statically reference @temporalio/workflow functions

@@ -106,31 +106,6 @@ // as well as Date and Math.random.

* Run a chunk of activation jobs
* @returns a boolean indicating whether job was processed or ignored
*/
function activate(activation, batchIndex) {
const activator = (0, global_attributes_1.getActivator)();
const intercept = (0, interceptors_1.composeInterceptors)(activator.interceptors.internals, 'activate', ({ activation, batchIndex }) => {
if (batchIndex === 0) {
if (!activation.jobs) {
throw new TypeError('Got activation with no jobs');
}
if (activation.timestamp != null) {
// timestamp will not be updated for activation that contain only queries
activator.now = (0, time_1.tsToMs)(activation.timestamp);
}
activator.addKnownFlags(activation.availableInternalFlags ?? []);
// The Rust Core ensures that these activation fields are not null
activator.mutateWorkflowInfo((info) => ({
...info,
historyLength: activation.historyLength,
// Exact truncation for multi-petabyte histories
// historySize === 0 means WFT was generated by pre-1.20.0 server, and the history size is unknown
historySize: activation.historySizeBytes?.toNumber() || 0,
continueAsNewSuggested: activation.continueAsNewSuggested ?? false,
currentBuildId: activation.buildIdForCurrentTask ?? undefined,
unsafe: {
...info.unsafe,
isReplaying: activation.isReplaying ?? false,
},
}));
}
const intercept = (0, interceptors_1.composeInterceptors)(activator.interceptors.internals, 'activate', ({ activation }) => {
// Cast from the interface to the class which has the `variant` attribute.

@@ -140,25 +115,13 @@ // This is safe because we know that activation is a proto class.

for (const job of jobs) {
if (job.variant === undefined) {
if (job.variant === undefined)
throw new TypeError('Expected job.variant to be defined');
}
const variant = job[job.variant];
if (!variant) {
if (!variant)
throw new TypeError(`Expected job.${job.variant} to be set`);
}
// The only job that can be executed on a completed workflow is a query.
// We might get other jobs after completion for instance when a single
// activation contains multiple jobs and the first one completes the workflow.
if (activator.completed && job.variant !== 'queryWorkflow') {
return;
}
activator[job.variant](variant /* TS can't infer this type */);
if (shouldUnblockConditions(job)) {
if (job.variant !== 'queryWorkflow')
tryUnblockConditions();
}
}
});
intercept({
activation,
batchIndex,
});
intercept({ activation, batchIndex });
}

@@ -176,7 +139,9 @@ exports.activate = activate;

const intercept = (0, interceptors_1.composeInterceptors)(activator.interceptors.internals, 'concludeActivation', (input) => input);
const { info } = activator;
const activationCompletion = activator.concludeActivation();
const { commands } = intercept({ commands: activationCompletion.commands });
if (activator.completed) {
activator.warnIfUnfinishedHandlers();
}
return {
runId: info.runId,
runId: activator.info.runId,
successful: { ...activationCompletion, commands },

@@ -186,6 +151,2 @@ };

exports.concludeActivation = concludeActivation;
function getAndResetSinkCalls() {
return (0, global_attributes_1.getActivator)().getAndResetSinkCalls();
}
exports.getAndResetSinkCalls = getAndResetSinkCalls;
/**

@@ -215,12 +176,6 @@ * Loop through all blocked conditions, evaluate and unblock if possible.

exports.tryUnblockConditions = tryUnblockConditions;
/**
* Predicate used to prevent triggering conditions for non-query and non-patch jobs.
*/
function shouldUnblockConditions(job) {
return !job.queryWorkflow && !job.notifyHasPatch;
}
exports.shouldUnblockConditions = shouldUnblockConditions;
function dispose() {
const dispose = (0, interceptors_1.composeInterceptors)((0, global_attributes_1.getActivator)().interceptors.internals, 'dispose', async () => {
(0, cancellation_scope_1.disableStorage)();
(0, update_scope_1.disableUpdateStorage)();
});

@@ -227,0 +182,0 @@ dispose({});

75

lib/workflow.d.ts
import { ActivityFunction, ActivityOptions, LocalActivityOptions, QueryDefinition, SearchAttributes, SignalDefinition, UntypedActivities, UpdateDefinition, WithWorkflowArgs, Workflow, WorkflowResultType, WorkflowReturnType } from '@temporalio/common';
import { Duration } from '@temporalio/common/lib/time';
import { temporal } from '@temporalio/proto';
import { ChildWorkflowOptions, ChildWorkflowOptionsWithDefaults, ContinueAsNewOptions, DefaultSignalHandler, EnhancedStackTrace, Handler, QueryHandlerOptions, SignalHandlerOptions, UpdateHandlerOptions, WorkflowInfo } from './interfaces';
import { ChildWorkflowOptions, ChildWorkflowOptionsWithDefaults, ContinueAsNewOptions, DefaultSignalHandler, EnhancedStackTrace, Handler, QueryHandlerOptions, SignalHandlerOptions, UpdateHandlerOptions, WorkflowInfo, UpdateInfo } from './interfaces';
import { ChildWorkflowHandle, ExternalWorkflowHandle } from './workflow-handle';
/**
* Adds default values to `workflowId` and `workflowIdReusePolicy` to given workflow options.
* Adds default values of `workflowId` and `cancellationType` to given workflow options.
*/

@@ -239,2 +239,11 @@ export declare function addDefaultWorkflowOptions<T extends Workflow>(opts: WithWorkflowArgs<T, ChildWorkflowOptions>): ChildWorkflowOptionsWithDefaults;

/**
* Get information about the current update if any.
*
* @return Info for the current update handler the code calling this is executing
* within if any.
*
* @experimental
*/
export declare function currentUpdateInfo(): UpdateInfo | undefined;
/**
* Returns whether or not code is executing in workflow context

@@ -327,4 +336,4 @@ */

*
* Definitions are used to register handler in the Workflow via {@link setHandler} and to update Workflows using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* Definitions can be reused in multiple Workflows.
* A definition is used to register a handler in the Workflow via {@link setHandler} and to update a Workflow using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* A definition can be reused in multiple Workflows.
*/

@@ -335,4 +344,4 @@ export declare function defineUpdate<Ret, Args extends any[] = [], Name extends string = string>(name: Name): UpdateDefinition<Ret, Args, Name>;

*
* Definitions are used to register handler in the Workflow via {@link setHandler} and to signal Workflows using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* Definitions can be reused in multiple Workflows.
* A definition is used to register a handler in the Workflow via {@link setHandler} and to signal a Workflow using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* A definition can be reused in multiple Workflows.
*/

@@ -343,4 +352,4 @@ export declare function defineSignal<Args extends any[] = [], Name extends string = string>(name: Name): SignalDefinition<Args, Name>;

*
* Definitions are used to register handler in the Workflow via {@link setHandler} and to query Workflows using a {@link WorkflowHandle}.
* Definitions can be reused in multiple Workflows.
* A definition is used to register a handler in the Workflow via {@link setHandler} and to query a Workflow using a {@link WorkflowHandle}.
* A definition can be reused in multiple Workflows.
*/

@@ -400,4 +409,54 @@ export declare function defineQuery<Ret, Args extends any[] = [], Name extends string = string>(name: Name): QueryDefinition<Ret, Args, Name>;

export declare function upsertSearchAttributes(searchAttributes: SearchAttributes): void;
/**
* Updates this Workflow's Memos by merging the provided `memo` with existing
* Memos (as returned by `workflowInfo().memo`).
*
* New memo is merged by replacing properties of the same name _at the first
* level only_. Setting a property to value `undefined` or `null` clears that
* key from the Memo.
*
* For example:
*
* ```ts
* upsertMemo({
* key1: value,
* key3: { subkey1: value }
* key4: value,
* });
* upsertMemo({
* key2: value
* key3: { subkey2: value }
* key4: undefined,
* });
* ```
*
* would result in the Workflow having these Memo:
*
* ```ts
* {
* key1: value,
* key2: value,
* key3: { subkey2: value } // Note this object was completely replaced
* // Note that key4 was completely removed
* }
* ```
*
* @param memo The Record to merge.
*/
export declare function upsertMemo(memo: Record<string, unknown>): void;
/**
* Whether update and signal handlers have finished executing.
*
* Consider waiting on this condition before workflow return or continue-as-new, to prevent
* interruption of in-progress handlers by workflow exit:
*
* ```ts
* await workflow.condition(workflow.allHandlersFinished)
* ```
*
* @returns true if there are no in-progress update or signal handler executions.
*/
export declare function allHandlersFinished(): boolean;
export declare const stackTraceQuery: QueryDefinition<string, [], string>;
export declare const enhancedStackTraceQuery: QueryDefinition<EnhancedStackTrace, [], string>;
export declare const workflowMetadataQuery: QueryDefinition<temporal.api.sdk.v1.IWorkflowMetadata, [], string>;

119

lib/workflow.js
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.workflowMetadataQuery = exports.enhancedStackTraceQuery = exports.stackTraceQuery = exports.upsertSearchAttributes = exports.setDefaultSignalHandler = exports.setHandler = exports.defineQuery = exports.defineSignal = exports.defineUpdate = exports.condition = exports.deprecatePatch = exports.patched = exports.uuid4 = exports.continueAsNew = exports.makeContinueAsNewFunc = exports.inWorkflowContext = exports.workflowInfo = exports.executeChild = exports.startChild = exports.getExternalWorkflowHandle = exports.proxyLocalActivities = exports.proxyActivities = exports.NotAnActivityMethod = exports.scheduleLocalActivity = exports.scheduleActivity = exports.sleep = exports.addDefaultWorkflowOptions = void 0;
exports.workflowMetadataQuery = exports.enhancedStackTraceQuery = exports.stackTraceQuery = exports.allHandlersFinished = exports.upsertMemo = exports.upsertSearchAttributes = exports.setDefaultSignalHandler = exports.setHandler = exports.defineQuery = exports.defineSignal = exports.defineUpdate = exports.condition = exports.deprecatePatch = exports.patched = exports.uuid4 = exports.continueAsNew = exports.makeContinueAsNewFunc = exports.inWorkflowContext = exports.currentUpdateInfo = exports.workflowInfo = exports.executeChild = exports.startChild = exports.getExternalWorkflowHandle = exports.proxyLocalActivities = exports.proxyActivities = exports.NotAnActivityMethod = exports.scheduleLocalActivity = exports.scheduleActivity = exports.sleep = exports.addDefaultWorkflowOptions = void 0;
const common_1 = require("@temporalio/common");

@@ -9,2 +9,3 @@ const versioning_intent_enum_1 = require("@temporalio/common/lib/versioning-intent-enum");

const cancellation_scope_1 = require("./cancellation-scope");
const update_scope_1 = require("./update-scope");
const interfaces_1 = require("./interfaces");

@@ -17,3 +18,3 @@ const errors_1 = require("./errors");

/**
* Adds default values to `workflowId` and `workflowIdReusePolicy` to given workflow options.
* Adds default values of `workflowId` and `cancellationType` to given workflow options.
*/

@@ -601,2 +602,15 @@ function addDefaultWorkflowOptions(opts) {

/**
* Get information about the current update if any.
*
* @return Info for the current update handler the code calling this is executing
* within if any.
*
* @experimental
*/
function currentUpdateInfo() {
(0, global_attributes_1.assertInWorkflowContext)('Workflow.currentUpdateInfo(...) may only be used from a Workflow Execution.');
return update_scope_1.UpdateScope.current();
}
exports.currentUpdateInfo = currentUpdateInfo;
/**
* Returns whether or not code is executing in workflow context

@@ -783,4 +797,4 @@ */

*
* Definitions are used to register handler in the Workflow via {@link setHandler} and to update Workflows using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* Definitions can be reused in multiple Workflows.
* A definition is used to register a handler in the Workflow via {@link setHandler} and to update a Workflow using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* A definition can be reused in multiple Workflows.
*/

@@ -797,4 +811,4 @@ function defineUpdate(name) {

*
* Definitions are used to register handler in the Workflow via {@link setHandler} and to signal Workflows using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* Definitions can be reused in multiple Workflows.
* A definition is used to register a handler in the Workflow via {@link setHandler} and to signal a Workflow using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* A definition can be reused in multiple Workflows.
*/

@@ -811,4 +825,4 @@ function defineSignal(name) {

*
* Definitions are used to register handler in the Workflow via {@link setHandler} and to query Workflows using a {@link WorkflowHandle}.
* Definitions can be reused in multiple Workflows.
* A definition is used to register a handler in the Workflow via {@link setHandler} and to query a Workflow using a {@link WorkflowHandle}.
* A definition can be reused in multiple Workflows.
*/

@@ -853,3 +867,3 @@ function defineQuery(name) {

// the Signal/Update will be executed before Workflow code is executed. If it
// is not, then the Signal/Update calls is pushed to a buffer.
// is not, then the Signal/Update calls are pushed to a buffer.
//

@@ -902,3 +916,3 @@ // 2. On each call to setHandler for a given Signal/Update, we make a pass

//
// 5. If an Update has a validation function then it is executed immediately
// 4. If an Update has a validation function then it is executed immediately
// prior to the handler. (Note that the validation function is required to be

@@ -913,3 +927,4 @@ // synchronous).

const validator = updateOptions?.validator;
activator.updateHandlers.set(def.name, { handler, validator, description });
const unfinishedPolicy = updateOptions?.unfinishedPolicy ?? common_1.HandlerUnfinishedPolicy.WARN_AND_ABANDON;
activator.updateHandlers.set(def.name, { handler, validator, description, unfinishedPolicy });
activator.dispatchBufferedUpdates();

@@ -926,3 +941,5 @@ }

if (typeof handler === 'function') {
activator.signalHandlers.set(def.name, { handler: handler, description });
const signalOptions = options;
const unfinishedPolicy = signalOptions?.unfinishedPolicy ?? common_1.HandlerUnfinishedPolicy.WARN_AND_ABANDON;
activator.signalHandlers.set(def.name, { handler: handler, description, unfinishedPolicy });
activator.dispatchBufferedSignals();

@@ -1026,2 +1043,80 @@ }

exports.upsertSearchAttributes = upsertSearchAttributes;
/**
* Updates this Workflow's Memos by merging the provided `memo` with existing
* Memos (as returned by `workflowInfo().memo`).
*
* New memo is merged by replacing properties of the same name _at the first
* level only_. Setting a property to value `undefined` or `null` clears that
* key from the Memo.
*
* For example:
*
* ```ts
* upsertMemo({
* key1: value,
* key3: { subkey1: value }
* key4: value,
* });
* upsertMemo({
* key2: value
* key3: { subkey2: value }
* key4: undefined,
* });
* ```
*
* would result in the Workflow having these Memo:
*
* ```ts
* {
* key1: value,
* key2: value,
* key3: { subkey2: value } // Note this object was completely replaced
* // Note that key4 was completely removed
* }
* ```
*
* @param memo The Record to merge.
*/
function upsertMemo(memo) {
const activator = (0, global_attributes_1.assertInWorkflowContext)('Workflow.upsertMemo(...) may only be used from a Workflow Execution.');
if (memo == null) {
throw new Error('memo must be a non-null Record');
}
activator.pushCommand({
modifyWorkflowProperties: {
upsertedMemo: {
fields: (0, common_1.mapToPayloads)(activator.payloadConverter,
// Convert null to undefined
Object.fromEntries(Object.entries(memo).map(([k, v]) => [k, v ?? undefined]))),
},
},
});
activator.mutateWorkflowInfo((info) => {
return {
...info,
memo: Object.fromEntries(Object.entries({
...info.memo,
...memo,
}).filter(([_, v]) => v != null)),
};
});
}
exports.upsertMemo = upsertMemo;
/**
* Whether update and signal handlers have finished executing.
*
* Consider waiting on this condition before workflow return or continue-as-new, to prevent
* interruption of in-progress handlers by workflow exit:
*
* ```ts
* await workflow.condition(workflow.allHandlersFinished)
* ```
*
* @returns true if there are no in-progress update or signal handler executions.
*/
function allHandlersFinished() {
const activator = (0, global_attributes_1.assertInWorkflowContext)('allHandlersFinished() may only be used from a Workflow Execution.');
return activator.inProgressSignals.size === 0 && activator.inProgressUpdates.size === 0;
}
exports.allHandlersFinished = allHandlersFinished;
exports.stackTraceQuery = defineQuery('__stack_trace');

@@ -1028,0 +1123,0 @@ exports.enhancedStackTraceQuery = defineQuery('__enhanced_stack_trace');

8

package.json
{
"name": "@temporalio/workflow",
"version": "1.10.3",
"version": "1.11.0",
"description": "Temporal.io SDK Workflow sub-package",

@@ -25,4 +25,4 @@ "keywords": [

"dependencies": {
"@temporalio/common": "1.10.3",
"@temporalio/proto": "1.10.3"
"@temporalio/common": "1.11.0",
"@temporalio/proto": "1.11.0"
},

@@ -39,3 +39,3 @@ "devDependencies": {

],
"gitHead": "7832067434443145e1f0ddf07cb37b0b90158bfb"
"gitHead": "3d7dd521cadc2ebcc4989360eeb0bedacd4533ab"
}

21

src/flags.ts

@@ -19,3 +19,22 @@ export type SdkFlag = {

*/
NonCancellableScopesAreShieldedFromPropagation: defineFlag(1, false),
NonCancellableScopesAreShieldedFromPropagation: defineFlag(1, true),
/**
* Prior to 1.11.0, when processing a Workflow activation, the SDK would execute `notifyHasPatch`
* and `signalWorkflow` jobs in distinct phases, before other types of jobs. The primary reason
* behind that multi-phase algorithm was to avoid the possibility that a Workflow execution might
* complete before all incoming signals have been dispatched (at least to the point that the
* _synchronous_ part of the handler function has been executed).
*
* This flag replaces that multi-phase algorithm with a simpler one where jobs are simply sorted as
* `(signals and updates) -> others`, but without processing them as distinct batches (i.e. without
* leaving/reentering the VM context between each group, which automatically triggers the execution
* of all outstanding microtasks). That single-phase approach resolves a number of quirks of the
* former algorithm, and yet still satisfies to the original requirement of ensuring that every
* `signalWorkflow` jobs - and now `doUpdate` jobs as well - have been given a proper chance to
* execute before the Workflow main function might completes.
*
* @since Introduced in 1.11.0. This change is not rollback-safe.
*/
ProcessWorkflowActivationJobsAsSingleBatch: defineFlag(2, true),
} as const;

@@ -22,0 +41,0 @@

6

src/index.ts

@@ -95,7 +95,7 @@ /**

EnhancedStackTrace,
FileLocation,
FileSlice,
StackTraceFileLocation,
StackTraceFileSlice,
ParentClosePolicy,
ParentWorkflowInfo,
SDKInfo,
StackTraceSDKInfo,
StackTrace,

@@ -102,0 +102,0 @@ UnsafeWorkflowInfo,

55

src/interfaces.ts

@@ -6,2 +6,3 @@ import type { RawSourceMap } from 'source-map';

CommonWorkflowOptions,
HandlerUnfinishedPolicy,
SearchAttributes,

@@ -194,2 +195,19 @@ SignalDefinition,

/**
* Information about a workflow update.
*
* @experimental
*/
export interface UpdateInfo {
/**
* A workflow-unique identifier for this update.
*/
readonly id: string;
/**
* The update type name.
*/
readonly name: string;
}
export interface ParentWorkflowInfo {

@@ -370,3 +388,3 @@ workflowId: string;

export interface SDKInfo {
export interface StackTraceSDKInfo {
name: string;

@@ -379,11 +397,11 @@ version: string;

*/
export interface FileSlice {
export interface StackTraceFileSlice {
/**
* Only used possible to trim the file without breaking syntax highlighting.
*/
line_offset: number;
/**
* slice of a file with `\n` (newline) line terminator.
*/
content: string;
/**
* Only used possible to trim the file without breaking syntax highlighting.
*/
lineOffset: number;
}

@@ -394,3 +412,3 @@

*/
export interface FileLocation {
export interface StackTraceFileLocation {
/**

@@ -400,3 +418,3 @@ * Path to source file (absolute or relative).

*/
filePath?: string;
file_path?: string;
/**

@@ -414,7 +432,11 @@ * If possible, SDK should send this, required for displaying the code location.

*/
functionName?: string;
function_name?: string;
/**
* Flag to mark this as internal SDK code and hide by default in the UI.
*/
internal_code: boolean;
}
export interface StackTrace {
locations: FileLocation[];
locations: StackTraceFileLocation[];
}

@@ -426,3 +448,3 @@

export interface EnhancedStackTrace {
sdk: SDKInfo;
sdk: StackTraceSDKInfo;
/**

@@ -433,3 +455,3 @@ * Mapping of file path to file contents.

*/
sources: Record<string, FileSlice[]>;
sources: Record<string, StackTraceFileSlice[]>;
stacks: StackTrace[];

@@ -443,2 +465,3 @@ }

patches: string[];
sdkFlags: number[];
showStackTraceSources: boolean;

@@ -486,3 +509,3 @@ }

*/
export type SignalHandlerOptions = { description?: string };
export type SignalHandlerOptions = { description?: string; unfinishedPolicy?: HandlerUnfinishedPolicy };

@@ -492,3 +515,7 @@ /**

*/
export type UpdateHandlerOptions<Args extends any[]> = { validator?: UpdateValidator<Args>; description?: string };
export type UpdateHandlerOptions<Args extends any[]> = {
validator?: UpdateValidator<Args>;
description?: string;
unfinishedPolicy?: HandlerUnfinishedPolicy;
};

@@ -495,0 +522,0 @@ export interface ActivationCompletion {

280

src/internals.ts

@@ -9,2 +9,3 @@ import type { RawSourceMap } from 'source-map';

ensureTemporalFailure,
HandlerUnfinishedPolicy,
IllegalStateError,

@@ -19,2 +20,4 @@ TemporalFailure,

ApplicationFailure,
WorkflowUpdateType,
WorkflowUpdateValidatorType,
} from '@temporalio/common';

@@ -26,2 +29,3 @@ import { composeInterceptors } from '@temporalio/common/lib/interceptors';

import { RootCancellationScope } from './cancellation-scope';
import { UpdateScope } from './update-scope';
import { DeterminismViolationError, LocalActivityDoBackoff, isCancellation } from './errors';

@@ -32,6 +36,6 @@ import { QueryInput, SignalInput, UpdateInput, WorkflowExecuteInput, WorkflowInterceptors } from './interceptors';

DefaultSignalHandler,
SDKInfo,
FileSlice,
StackTraceSDKInfo,
StackTraceFileSlice,
EnhancedStackTrace,
FileLocation,
StackTraceFileLocation,
WorkflowInfo,

@@ -44,4 +48,4 @@ WorkflowCreateOptionsInternal,

import pkg from './pkg';
import { executeWithLifecycleLogging } from './logs';
import { SdkFlag, assertValidFlag } from './flags';
import { executeWithLifecycleLogging, log } from './logs';

@@ -58,3 +62,3 @@ enum StartChildWorkflowExecutionFailedCause {

formatted: string;
structured: FileLocation[];
structured: StackTraceFileLocation[];
}

@@ -72,2 +76,3 @@

resolve(val: unknown): unknown;
reject(reason: unknown): unknown;

@@ -78,2 +83,3 @@ }

fn(): boolean;
resolve(): void;

@@ -94,5 +100,27 @@ }

/**
* Information about an update or signal handler execution.
*/
interface MessageHandlerExecution {
name: string;
unfinishedPolicy: HandlerUnfinishedPolicy;
id?: string;
}
/**
* Keeps all of the Workflow runtime state like pending completions for activities and timers.
*
* Implements handlers for all workflow activation jobs.
*
* Note that most methods in this class are meant to be called only from within the VM.
*
* However, a few methods may be called directly from outside the VM (essentially from `vm-shared.ts`).
* These methods are specifically marked with a comment and require careful consideration, as the
* execution context may not properly reflect that of the target workflow execution (e.g.: with Reusable
* VMs, the `global` may not have been swapped to those of that workflow execution; the active microtask
* queue may be that of the thread/process, rather than the queue of that VM context; etc). Consequently,
* methods that are meant to be called from outside of the VM must not do any of the following:
*
* - Access any global variable;
* - Create Promise objects, use async/await, or otherwise schedule microtasks;
* - Call user-defined functions, including any form of interceptor.
*/

@@ -127,11 +155,2 @@ export class Activator implements ActivationHandler {

/**
* Holds buffered query calls until a handler is registered.
*
* **IMPORTANT** queries are only buffered until workflow is started.
* This is required because async interceptors might block workflow function invocation
* which delays query handler registration.
*/
protected readonly bufferedQueries = Array<coresdk.workflow_activation.IQueryWorkflow>();
/**
* Mapping of update name to handler and validator

@@ -147,2 +166,17 @@ */

/**
* Mapping of in-progress updates to handler execution information.
*/
readonly inProgressUpdates = new Map<string, MessageHandlerExecution>();
/**
* Mapping of in-progress signals to handler execution information.
*/
readonly inProgressSignals = new Map<number, MessageHandlerExecution>();
/**
* A sequence number providing unique identifiers for signal handler executions.
*/
protected signalHandlerExecutionSeq = 0;
/**
* A signal handler that catches calls for non-registered signal names.

@@ -189,15 +223,15 @@ */

const { sourceMap } = this;
const sdk: SDKInfo = { name: 'typescript', version: pkg.version };
const sdk: StackTraceSDKInfo = { name: 'typescript', version: pkg.version };
const stacks = this.getStackTraces().map(({ structured: locations }) => ({ locations }));
const sources: Record<string, FileSlice[]> = {};
const sources: Record<string, StackTraceFileSlice[]> = {};
if (this.showStackTraceSources) {
for (const { locations } of stacks) {
for (const { filePath } of locations) {
if (!filePath) continue;
const content = sourceMap?.sourcesContent?.[sourceMap?.sources.indexOf(filePath)];
for (const { file_path } of locations) {
if (!file_path) continue;
const content = sourceMap?.sourcesContent?.[sourceMap?.sources.indexOf(file_path)];
if (!content) continue;
sources[filePath] = [
sources[file_path] = [
{
line_offset: 0,
content,
lineOffset: 0,
},

@@ -233,3 +267,2 @@ ];

type: workflowType,
description: null, // For now, do not set the workflow description in the TS SDK.
queryDefinitions,

@@ -249,3 +282,7 @@ signalDefinitions,

*/
public readonly interceptors: Required<WorkflowInterceptors> = { inbound: [], outbound: [], internals: [] };
public readonly interceptors: Required<WorkflowInterceptors> = {
inbound: [],
outbound: [],
internals: [],
};

@@ -277,9 +314,2 @@ /**

/**
* This is tracked to allow buffering queries until a workflow function is called.
* TODO(bergundy): I don't think this makes sense since queries run last in an activation and must be responded to in
* the same activation.
*/
protected workflowFunctionWasCalled = false;
/**
* The next (incremental) sequence to assign when generating completable commands

@@ -300,2 +330,3 @@ */

* This is set every time the workflow executes an activation
* May be accessed and modified from outside the VM.
*/

@@ -311,2 +342,3 @@ now: number;

* Information about the current Workflow
* May be accessed from outside the VM.
*/

@@ -353,2 +385,3 @@ public info: WorkflowInfo;

getTimeOfDay,
sdkFlags,
randomnessSeed,

@@ -366,9 +399,11 @@ patches,

if (info.unsafe.isReplaying) {
for (const patchId of patches) {
this.notifyHasPatch({ patchId });
}
this.addKnownFlags(sdkFlags);
for (const patchId of patches) {
this.notifyHasPatch({ patchId });
}
}
/**
* May be invoked from outside the VM.
*/
mutateWorkflowInfo(fn: (info: WorkflowInfo) => WorkflowInfo): void {

@@ -400,2 +435,5 @@ this.info = fn(this.info);

/**
* May be invoked from outside the VM.
*/
getAndResetSinkCalls(): SinkCall[] {

@@ -413,4 +451,2 @@ const { sinkCalls } = this;

pushCommand(cmd: coresdk.workflow_commands.IWorkflowCommand, complete = false): void {
// Only query responses may be sent after completion
if (this.completed && !cmd.respondToQuery) return;
this.commands.push(cmd);

@@ -434,15 +470,3 @@ if (complete) {

}
let promise: Promise<any>;
try {
promise = workflow(...args);
} finally {
// Queries must be handled even if there was an exception when invoking the Workflow function.
this.workflowFunctionWasCalled = true;
// Empty the buffer
const buffer = this.bufferedQueries.splice(0);
for (const activation of buffer) {
this.queryWorkflow(activation);
}
}
return await promise;
return await workflow(...args);
}

@@ -578,7 +602,2 @@

public queryWorkflow(activation: coresdk.workflow_activation.IQueryWorkflow): void {
if (!this.workflowFunctionWasCalled) {
this.bufferedQueries.push(activation);
return;
}
const { queryType, queryId, headers } = activation;

@@ -616,3 +635,4 @@ if (!(queryType && queryId)) {

}
if (!this.updateHandlers.has(name)) {
const entry = this.updateHandlers.get(name);
if (!entry) {
this.bufferedUpdates.push(activation);

@@ -656,21 +676,27 @@ return;

// These are caught elsewhere and fail the corresponding activation.
let input: UpdateInput;
try {
if (runValidator && this.updateHandlers.get(name)?.validator) {
const validate = composeInterceptors(
this.interceptors.inbound,
'validateUpdate',
this.validateUpdateNextHandler.bind(this)
);
validate(makeInput());
const doUpdateImpl = async () => {
let input: UpdateInput;
try {
if (runValidator && entry.validator) {
const validate = composeInterceptors(
this.interceptors.inbound,
'validateUpdate',
this.validateUpdateNextHandler.bind(this, entry.validator)
);
validate(makeInput());
}
input = makeInput();
} catch (error) {
this.rejectUpdate(protocolInstanceId, error);
return;
}
input = makeInput();
} catch (error) {
this.rejectUpdate(protocolInstanceId, error);
return;
}
const execute = composeInterceptors(this.interceptors.inbound, 'handleUpdate', this.updateNextHandler.bind(this));
this.acceptUpdate(protocolInstanceId);
untrackPromise(
execute(input)
this.acceptUpdate(protocolInstanceId);
const execute = composeInterceptors(
this.interceptors.inbound,
'handleUpdate',
this.updateNextHandler.bind(this, entry.handler)
);
const { unfinishedPolicy } = entry;
this.inProgressUpdates.set(updateId, { name, unfinishedPolicy, id: updateId });
const res = execute(input)
.then((result) => this.completeUpdate(protocolInstanceId, result))

@@ -684,16 +710,14 @@ .catch((error) => {

})
);
.finally(() => this.inProgressUpdates.delete(updateId));
untrackPromise(res);
return res;
};
untrackPromise(UpdateScope.updateWithInfo(updateId, name, doUpdateImpl));
}
protected async updateNextHandler({ name, args }: UpdateInput): Promise<unknown> {
const entry = this.updateHandlers.get(name);
if (!entry) {
return Promise.reject(new IllegalStateError(`No registered update handler for update: ${name}`));
}
const { handler } = entry;
protected async updateNextHandler(handler: WorkflowUpdateType, { args }: UpdateInput): Promise<unknown> {
return await handler(...args);
}
protected validateUpdateNextHandler({ name, args }: UpdateInput): void {
const { validator } = this.updateHandlers.get(name) ?? {};
protected validateUpdateNextHandler(validator: WorkflowUpdateValidatorType | undefined, { args }: UpdateInput): void {
if (validator) {

@@ -752,2 +776,10 @@ validator(...args);

// If we fall through to the default signal handler then the unfinished
// policy is WARN_AND_ABANDON; users currently have no way to silence any
// ensuing warnings.
const unfinishedPolicy =
this.signalHandlers.get(signalName)?.unfinishedPolicy ?? HandlerUnfinishedPolicy.WARN_AND_ABANDON;
const signalExecutionNum = this.signalHandlerExecutionSeq++;
this.inProgressSignals.set(signalExecutionNum, { name: signalName, unfinishedPolicy });
const execute = composeInterceptors(

@@ -762,3 +794,5 @@ this.interceptors.inbound,

headers: headers ?? {},
}).catch(this.handleWorkflowFailure.bind(this));
})
.catch(this.handleWorkflowFailure.bind(this))
.finally(() => this.inProgressSignals.delete(signalExecutionNum));
}

@@ -802,2 +836,20 @@

public warnIfUnfinishedHandlers(): void {
const getWarnable = (handlerExecutions: Iterable<MessageHandlerExecution>): MessageHandlerExecution[] => {
return Array.from(handlerExecutions).filter(
(ex) => ex.unfinishedPolicy === HandlerUnfinishedPolicy.WARN_AND_ABANDON
);
};
const warnableUpdates = getWarnable(this.inProgressUpdates.values());
if (warnableUpdates.length > 0) {
log.warn(makeUnfinishedUpdateHandlerMessage(warnableUpdates));
}
const warnableSignals = getWarnable(this.inProgressSignals.values());
if (warnableSignals.length > 0) {
log.warn(makeUnfinishedSignalHandlerMessage(warnableSignals));
}
}
public updateRandomSeed(activation: coresdk.workflow_activation.IUpdateRandomSeed): void {

@@ -811,5 +863,5 @@ if (!activation.randomnessSeed) {

public notifyHasPatch(activation: coresdk.workflow_activation.INotifyHasPatch): void {
if (!activation.patchId) {
throw new TypeError('Notify has patch missing patch name');
}
if (!this.info.unsafe.isReplaying)
throw new IllegalStateError('Unexpected notifyHasPatch job on non-replay activation');
if (!activation.patchId) throw new TypeError('notifyHasPatch missing patch id');
this.knownPresentPatches.add(activation.patchId);

@@ -834,3 +886,6 @@ }

// Called early while handling an activation to register known flags
/**
* Called early while handling an activation to register known flags.
* May be invoked from outside the VM.
*/
public addKnownFlags(flags: number[]): void {

@@ -843,2 +898,7 @@ for (const flag of flags) {

/**
* Check if a flag is known to the Workflow Execution; if not, enable the flag if workflow
* is not replaying and the flag is configured to be enabled by default.
* May be invoked from outside the VM.
*/
public hasFlag(flag: SdkFlag): boolean {

@@ -874,3 +934,6 @@ if (this.knownFlags.has(flag.id)) {

}
// Fail the workflow. We do not want to issue unfinishedHandlers warnings. To achieve that, we
// mark all handlers as completed now.
this.inProgressSignals.clear();
this.inProgressUpdates.clear();
this.pushCommand(

@@ -966,1 +1029,42 @@ {

}
function makeUnfinishedUpdateHandlerMessage(handlerExecutions: MessageHandlerExecution[]): string {
const message = `
[TMPRL1102] Workflow finished while an update handler was still running. This may have interrupted work that the
update handler was doing, and the client that sent the update will receive a 'workflow execution
already completed' RPCError instead of the update result. You can wait for all update and signal
handlers to complete by using \`await workflow.condition(workflow.allHandlersFinished)\`.
Alternatively, if both you and the clients sending the update are okay with interrupting running handlers
when the workflow finishes, and causing clients to receive errors, then you can disable this warning by
passing an option when setting the handler:
\`workflow.setHandler(myUpdate, myUpdateHandler, {unfinishedPolicy: HandlerUnfinishedPolicy.ABANDON});\`.`
.replace(/\n/g, ' ')
.trim();
return `${message} The following updates were unfinished (and warnings were not disabled for their handler): ${JSON.stringify(
handlerExecutions.map((ex) => ({ name: ex.name, id: ex.id }))
)}`;
}
function makeUnfinishedSignalHandlerMessage(handlerExecutions: MessageHandlerExecution[]): string {
const message = `
[TMPRL1102] Workflow finished while a signal handler was still running. This may have interrupted work that the
signal handler was doing. You can wait for all update and signal handlers to complete by using
\`await workflow.condition(workflow.allHandlersFinished)\`. Alternatively, if both you and the
clients sending the update are okay with interrupting running handlers when the workflow finishes,
then you can disable this warning by passing an option when setting the handler:
\`workflow.setHandler(mySignal, mySignalHandler, {unfinishedPolicy: HandlerUnfinishedPolicy.ABANDON});\`.`
.replace(/\n/g, ' ')
.trim();
const names = new Map<string, number>();
for (const ex of handlerExecutions) {
const count = names.get(ex.name) || 0;
names.set(ex.name, count + 1);
}
return `${message} The following signals were unfinished (and warnings were not disabled for their handler): ${JSON.stringify(
Array.from(names.entries()).map(([name, count]) => ({ name, count }))
)}`;
}

79

src/worker-interface.ts

@@ -7,6 +7,6 @@ /**

import { IllegalStateError } from '@temporalio/common';
import { tsToMs } from '@temporalio/common/lib/time';
import { composeInterceptors } from '@temporalio/common/lib/interceptors';
import { coresdk } from '@temporalio/proto';
import { disableStorage } from './cancellation-scope';
import { disableUpdateStorage } from './update-scope';
import { WorkflowInterceptorsFactory } from './interceptors';

@@ -16,3 +16,2 @@ import { WorkflowCreateOptionsInternal } from './interfaces';

import { setActivatorUntyped, getActivator } from './global-attributes';
import { type SinkCall } from './sinks';

@@ -38,3 +37,3 @@ // Export the type for use on the "worker" side

});
// There's on activator per workflow instance, set it globally on the context.
// There's one activator per workflow instance, set it globally on the context.
// We do this before importing any user code so user code can statically reference @temporalio/workflow functions

@@ -116,33 +115,6 @@ // as well as Date and Math.random.

* Run a chunk of activation jobs
* @returns a boolean indicating whether job was processed or ignored
*/
export function activate(activation: coresdk.workflow_activation.WorkflowActivation, batchIndex: number): void {
export function activate(activation: coresdk.workflow_activation.IWorkflowActivation, batchIndex: number): void {
const activator = getActivator();
const intercept = composeInterceptors(activator.interceptors.internals, 'activate', ({ activation, batchIndex }) => {
if (batchIndex === 0) {
if (!activation.jobs) {
throw new TypeError('Got activation with no jobs');
}
if (activation.timestamp != null) {
// timestamp will not be updated for activation that contain only queries
activator.now = tsToMs(activation.timestamp);
}
activator.addKnownFlags(activation.availableInternalFlags ?? []);
// The Rust Core ensures that these activation fields are not null
activator.mutateWorkflowInfo((info) => ({
...info,
historyLength: activation.historyLength as number,
// Exact truncation for multi-petabyte histories
// historySize === 0 means WFT was generated by pre-1.20.0 server, and the history size is unknown
historySize: activation.historySizeBytes?.toNumber() || 0,
continueAsNewSuggested: activation.continueAsNewSuggested ?? false,
currentBuildId: activation.buildIdForCurrentTask ?? undefined,
unsafe: {
...info.unsafe,
isReplaying: activation.isReplaying ?? false,
},
}));
}
const intercept = composeInterceptors(activator.interceptors.internals, 'activate', ({ activation }) => {
// Cast from the interface to the class which has the `variant` attribute.

@@ -153,26 +125,13 @@ // This is safe because we know that activation is a proto class.

for (const job of jobs) {
if (job.variant === undefined) {
throw new TypeError('Expected job.variant to be defined');
}
if (job.variant === undefined) throw new TypeError('Expected job.variant to be defined');
const variant = job[job.variant];
if (!variant) {
throw new TypeError(`Expected job.${job.variant} to be set`);
}
// The only job that can be executed on a completed workflow is a query.
// We might get other jobs after completion for instance when a single
// activation contains multiple jobs and the first one completes the workflow.
if (activator.completed && job.variant !== 'queryWorkflow') {
return;
}
if (!variant) throw new TypeError(`Expected job.${job.variant} to be set`);
activator[job.variant](variant as any /* TS can't infer this type */);
if (shouldUnblockConditions(job)) {
tryUnblockConditions();
}
if (job.variant !== 'queryWorkflow') tryUnblockConditions();
}
});
intercept({
activation,
batchIndex,
});
intercept({ activation, batchIndex });
}

@@ -190,8 +149,10 @@

const intercept = composeInterceptors(activator.interceptors.internals, 'concludeActivation', (input) => input);
const { info } = activator;
const activationCompletion = activator.concludeActivation();
const { commands } = intercept({ commands: activationCompletion.commands });
if (activator.completed) {
activator.warnIfUnfinishedHandlers();
}
return {
runId: info.runId,
runId: activator.info.runId,
successful: { ...activationCompletion, commands },

@@ -201,6 +162,2 @@ };

export function getAndResetSinkCalls(): SinkCall[] {
return getActivator().getAndResetSinkCalls();
}
/**

@@ -230,14 +187,8 @@ * Loop through all blocked conditions, evaluate and unblock if possible.

/**
* Predicate used to prevent triggering conditions for non-query and non-patch jobs.
*/
export function shouldUnblockConditions(job: coresdk.workflow_activation.IWorkflowActivationJob): boolean {
return !job.queryWorkflow && !job.notifyHasPatch;
}
export function dispose(): void {
const dispose = composeInterceptors(getActivator().interceptors.internals, 'dispose', async () => {
disableStorage();
disableUpdateStorage();
});
dispose({});
}

127

src/workflow.ts

@@ -6,2 +6,3 @@ import {

extractWorkflowType,
HandlerUnfinishedPolicy,
LocalActivityOptions,

@@ -27,2 +28,3 @@ mapToPayloads,

import { CancellationScope, registerSleepImplementation } from './cancellation-scope';
import { UpdateScope } from './update-scope';
import {

@@ -48,2 +50,3 @@ ActivityInput,

WorkflowInfo,
UpdateInfo,
} from './interfaces';

@@ -59,3 +62,3 @@ import { LocalActivityDoBackoff } from './errors';

/**
* Adds default values to `workflowId` and `workflowIdReusePolicy` to given workflow options.
* Adds default values of `workflowId` and `cancellationType` to given workflow options.
*/

@@ -875,2 +878,15 @@ export function addDefaultWorkflowOptions<T extends Workflow>(

/**
* Get information about the current update if any.
*
* @return Info for the current update handler the code calling this is executing
* within if any.
*
* @experimental
*/
export function currentUpdateInfo(): UpdateInfo | undefined {
assertInWorkflowContext('Workflow.currentUpdateInfo(...) may only be used from a Workflow Execution.');
return UpdateScope.current();
}
/**
* Returns whether or not code is executing in workflow context

@@ -1088,4 +1104,4 @@ */

*
* Definitions are used to register handler in the Workflow via {@link setHandler} and to update Workflows using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* Definitions can be reused in multiple Workflows.
* A definition is used to register a handler in the Workflow via {@link setHandler} and to update a Workflow using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* A definition can be reused in multiple Workflows.
*/

@@ -1104,4 +1120,4 @@ export function defineUpdate<Ret, Args extends any[] = [], Name extends string = string>(

*
* Definitions are used to register handler in the Workflow via {@link setHandler} and to signal Workflows using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* Definitions can be reused in multiple Workflows.
* A definition is used to register a handler in the Workflow via {@link setHandler} and to signal a Workflow using a {@link WorkflowHandle}, {@link ChildWorkflowHandle} or {@link ExternalWorkflowHandle}.
* A definition can be reused in multiple Workflows.
*/

@@ -1120,4 +1136,4 @@ export function defineSignal<Args extends any[] = [], Name extends string = string>(

*
* Definitions are used to register handler in the Workflow via {@link setHandler} and to query Workflows using a {@link WorkflowHandle}.
* Definitions can be reused in multiple Workflows.
* A definition is used to register a handler in the Workflow via {@link setHandler} and to query a Workflow using a {@link WorkflowHandle}.
* A definition can be reused in multiple Workflows.
*/

@@ -1189,3 +1205,3 @@ export function defineQuery<Ret, Args extends any[] = [], Name extends string = string>(

// the Signal/Update will be executed before Workflow code is executed. If it
// is not, then the Signal/Update calls is pushed to a buffer.
// is not, then the Signal/Update calls are pushed to a buffer.
//

@@ -1238,3 +1254,3 @@ // 2. On each call to setHandler for a given Signal/Update, we make a pass

//
// 5. If an Update has a validation function then it is executed immediately
// 4. If an Update has a validation function then it is executed immediately
// prior to the handler. (Note that the validation function is required to be

@@ -1256,4 +1272,6 @@ // synchronous).

const updateOptions = options as UpdateHandlerOptions<Args> | undefined;
const validator = updateOptions?.validator as WorkflowUpdateValidatorType | undefined;
activator.updateHandlers.set(def.name, { handler, validator, description });
const unfinishedPolicy = updateOptions?.unfinishedPolicy ?? HandlerUnfinishedPolicy.WARN_AND_ABANDON;
activator.updateHandlers.set(def.name, { handler, validator, description, unfinishedPolicy });
activator.dispatchBufferedUpdates();

@@ -1267,3 +1285,5 @@ } else if (handler == null) {

if (typeof handler === 'function') {
activator.signalHandlers.set(def.name, { handler: handler as any, description });
const signalOptions = options as SignalHandlerOptions | undefined;
const unfinishedPolicy = signalOptions?.unfinishedPolicy ?? HandlerUnfinishedPolicy.WARN_AND_ABANDON;
activator.signalHandlers.set(def.name, { handler: handler as any, description, unfinishedPolicy });
activator.dispatchBufferedSignals();

@@ -1366,4 +1386,89 @@ } else if (handler == null) {

/**
* Updates this Workflow's Memos by merging the provided `memo` with existing
* Memos (as returned by `workflowInfo().memo`).
*
* New memo is merged by replacing properties of the same name _at the first
* level only_. Setting a property to value `undefined` or `null` clears that
* key from the Memo.
*
* For example:
*
* ```ts
* upsertMemo({
* key1: value,
* key3: { subkey1: value }
* key4: value,
* });
* upsertMemo({
* key2: value
* key3: { subkey2: value }
* key4: undefined,
* });
* ```
*
* would result in the Workflow having these Memo:
*
* ```ts
* {
* key1: value,
* key2: value,
* key3: { subkey2: value } // Note this object was completely replaced
* // Note that key4 was completely removed
* }
* ```
*
* @param memo The Record to merge.
*/
export function upsertMemo(memo: Record<string, unknown>): void {
const activator = assertInWorkflowContext('Workflow.upsertMemo(...) may only be used from a Workflow Execution.');
if (memo == null) {
throw new Error('memo must be a non-null Record');
}
activator.pushCommand({
modifyWorkflowProperties: {
upsertedMemo: {
fields: mapToPayloads(
activator.payloadConverter,
// Convert null to undefined
Object.fromEntries(Object.entries(memo).map(([k, v]) => [k, v ?? undefined]))
),
},
},
});
activator.mutateWorkflowInfo((info: WorkflowInfo): WorkflowInfo => {
return {
...info,
memo: Object.fromEntries(
Object.entries({
...info.memo,
...memo,
}).filter(([_, v]) => v != null)
),
};
});
}
/**
* Whether update and signal handlers have finished executing.
*
* Consider waiting on this condition before workflow return or continue-as-new, to prevent
* interruption of in-progress handlers by workflow exit:
*
* ```ts
* await workflow.condition(workflow.allHandlersFinished)
* ```
*
* @returns true if there are no in-progress update or signal handler executions.
*/
export function allHandlersFinished(): boolean {
const activator = assertInWorkflowContext('allHandlersFinished() may only be used from a Workflow Execution.');
return activator.inProgressSignals.size === 0 && activator.inProgressUpdates.size === 0;
}
export const stackTraceQuery = defineQuery<string>('__stack_trace');
export const enhancedStackTraceQuery = defineQuery<EnhancedStackTrace>('__enhanced_stack_trace');
export const workflowMetadataQuery = defineQuery<temporal.api.sdk.v1.IWorkflowMetadata>('__temporal_workflow_metadata');
lib/flags.js.map

Sorry, the diff of this file is not supported yet

lib/interfaces.js.map

Sorry, the diff of this file is not supported yet

lib/internals.js.map

Sorry, the diff of this file is not supported yet

lib/worker-interface.js.map

Sorry, the diff of this file is not supported yet

lib/workflow.js.map

Sorry, the diff of this file is not supported yet

SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap
  • Changelog

About

Packages

npm

Go

Maven

PyPI

Rubygems

Stay in touch

Get open source security insights delivered straight into your inbox.

  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc