precheck signal

Author: grutt

sdks/typescript/src/clients/listeners/durable-listener/pooled-durable-listener-client.ts

@@ -8,20 +8,14 @@ import {
 } from '@hatchet/protoc/v1/dispatcher';
 import { isAbortError } from 'abort-controller-x';
 import sleep from '@hatchet/util/sleep';
+import { createAbortError } from '@hatchet/util/abort-error';
 import {
   DurableEventListenerConditions,
   SleepMatchCondition,
   UserEventMatchCondition,
 } from '@hatchet/protoc/v1/shared/condition';
 import { DurableListenerClient } from './durable-listener-client';
 
-function createAbortError(message = 'Operation aborted'): Error {
-  const err: any = new Error(message);
-  err.name = 'AbortError';
-  err.code = 'ABORT_ERR';
-  return err as Error;
-}
-
 export class DurableEventStreamable {
   listener: AsyncIterable<DurableEvent>;
   taskId: string;

sdks/typescript/src/clients/listeners/run-listener/pooled-child-listener-client.ts

@@ -7,15 +7,9 @@ import {
 } from '@hatchet/protoc/dispatcher';
 import { isAbortError } from 'abort-controller-x';
 import sleep from '@hatchet/util/sleep';
+import { createAbortError } from '@hatchet/util/abort-error';
 import { RunListenerClient } from './child-listener-client';
 
-function createAbortError(message = 'Operation aborted'): Error {
-  const err: any = new Error(message);
-  err.name = 'AbortError';
-  err.code = 'ABORT_ERR';
-  return err as Error;
-}
-
 export class Streamable {
   listener: AsyncIterable<WorkflowRunEvent>;
   id: string;

sdks/typescript/src/clients/worker/worker.ts

@@ -520,7 +520,7 @@ export class V0Worker {
       if (future) {
         future.promise.catch(() => {
           this.logger.info(
-            taskRunLog(action.taskName, action.taskRunExternalId, 'cancellation completed')
+            taskRunLog(action.taskName, action.taskRunExternalId, 'cancelled')
           );
         });
         future.cancel('Cancelled by worker');
@@ -529,7 +529,7 @@ export class V0Worker {
     } catch (e: any) {
       // Expected: the promise rejects when cancelled
       this.logger.debug(
-        taskRunLog(action.taskName, action.taskRunExternalId, 'cancellation completed')
+        taskRunLog(action.taskName, action.taskRunExternalId, 'cancelled')
       );
     } finally {
       delete this.futures[createActionKey(action)];

sdks/typescript/src/util/abort-error.ts

@@ -0,0 +1,6 @@
+export function createAbortError(message = 'Operation aborted'): Error {
+  const err: any = new Error(message);
+  err.name = 'AbortError';
+  err.code = 'ABORT_ERR';
+  return err as Error;
+}

sdks/typescript/src/v1/client/worker/context.ts

@@ -21,6 +21,7 @@ import { Action as ConditionAction } from '@hatchet/protoc/v1/shared/condition';
 import { HatchetClient } from '@hatchet/v1';
 import { ContextWorker, NextStep } from '@hatchet/step';
 import { applyNamespace } from '@hatchet/util/apply-namespace';
+import { createAbortError } from '@hatchet/util/abort-error';
 import { V1Worker } from './worker-internal';
 import { Duration } from '../duration';
 
@@ -90,6 +91,12 @@ export class Context<T, K = {}> {
     return this.controller.signal.aborted;
   }
 
+  protected throwIfCancelled(): void {
+    if (this.abortController.signal.aborted) {
+      throw createAbortError('Operation cancelled by AbortSignal');
+    }
+  }
+
   async cancel() {
     await this.v1.runs.cancel({
       ids: [this.action.taskRunExternalId],
@@ -359,6 +366,8 @@ export class Context<T, K = {}> {
   }
 
   private spawnOptions(workflow: string | Workflow | WorkflowV1<any, any>, options?: ChildRunOpts) {
+    this.throwIfCancelled();
+
     let workflowName: string;
 
     if (typeof workflow === 'string') {
@@ -410,6 +419,7 @@ export class Context<T, K = {}> {
       options?: ChildRunOpts;
     }>
   ) {
+    this.throwIfCancelled();
     const workflows: Parameters<typeof this.v1.admin.runWorkflows<Q, P>>[0] = children.map(
       (child) => {
         const { workflowName, opts } = this.spawnOptions(child.workflow, child.options);
@@ -595,6 +605,7 @@ export class Context<T, K = {}> {
       options?: ChildRunOpts;
     }>
   ): Promise<WorkflowRunRef<P>[]> {
+    this.throwIfCancelled();
     const { workflowRunId, taskRunExternalId } = this.action;
 
     const workflowRuns = workflows.map(({ workflow, input, options }) => {
@@ -677,6 +688,7 @@ export class Context<T, K = {}> {
     input: Q,
     options?: ChildRunOpts
   ): Promise<WorkflowRunRef<P>> {
+    this.throwIfCancelled();
     const { workflowRunId, taskRunExternalId } = this.action;
 
     let workflowName: string = '';
@@ -747,6 +759,7 @@ export class DurableContext<T, K = {}> extends Context<T, K> {
    * @returns A promise that resolves with the event that satisfied the conditions.
    */
   async waitFor(conditions: Conditions | Conditions[]): Promise<Record<string, any>> {
+    this.throwIfCancelled();
     const pbConditions = conditionsToPb(Render(ConditionAction.CREATE, conditions));
 
     // eslint-disable-next-line no-plusplus

sdks/typescript/src/v1/client/worker/worker-internal.ts

@@ -886,7 +886,7 @@ export class V1Worker {
         ]);
 
         if (winner === 'done') {
-          this.logger.info(taskRunLog(taskName, taskRunExternalId, 'cancellation completed'));
+          this.logger.info(taskRunLog(taskName, taskRunExternalId, 'cancelled'));
         } else {
           const totalElapsedMs = Date.now() - start;
           this.logger.warn(

sdks/typescript/src/v1/declaration.ts

@@ -25,9 +25,26 @@ import { MetricsClient } from './client/features/metrics';
 import { InputType, OutputType, UnknownInputType, JsonObject } from './types';
 import { Context, DurableContext } from './client/worker/context';
 import { parentRunContextManager } from './parent-run-context-vars';
+import { createAbortError } from '@hatchet/util/abort-error';
 
 const UNBOUND_ERR = new Error('workflow unbound to hatchet client, hint: use client.run instead');
 
+function throwIfAborted(signal: AbortSignal | undefined): void {
+  if (!signal?.aborted) {
+    return;
+  }
+
+  const reason = (signal as any).reason;
+
+  if (reason instanceof Error) {
+    throw reason;
+  }
+
+  throw createAbortError(
+    typeof reason === 'string' && reason.length > 0 ? reason : 'Operation cancelled by AbortSignal'
+  );
+}
+
 // eslint-disable-next-line no-shadow
 export enum Priority {
   LOW = 1,
@@ -312,7 +329,6 @@ export class BaseWorkflowDeclaration<
 
     // set the parent run context
     const parentRunContext = parentRunContextManager.getContext();
-    parentRunContextManager.incrementChildIndex(Array.isArray(input) ? input.length : 1);
 
     if (!parentRunContext && (options?.childKey || options?.sticky)) {
       this.client.admin.logger.warn(
@@ -322,6 +338,12 @@ export class BaseWorkflowDeclaration<
 
     const inheritedSignal = parentRunContext?.signal;
 
+    // Precheck: if we're being called from a cancelled parent task, do not enqueue more work.
+    // The signal is inherited from the parent task's `ctx.abortController.signal`.
+    throwIfAborted(inheritedSignal);
+
+    parentRunContextManager.incrementChildIndex(Array.isArray(input) ? input.length : 1);
+
     const runOpts = {
       ...(options ?? {}),
       parentId: parentRunContext?.parentId,
@@ -438,6 +460,9 @@ export class BaseWorkflowDeclaration<
       throw UNBOUND_ERR;
     }
 
+    // If called from within a cancelled parent task, do not enqueue scheduled work.
+    throwIfAborted(parentRunContextManager.getContext()?.signal);
+
     const scheduled = this.client.scheduled.create(this.definition.name, {
       triggerAt: enqueueAt,
       input: input as JsonObject,
@@ -480,6 +505,9 @@ export class BaseWorkflowDeclaration<
       throw UNBOUND_ERR;
     }
 
+    // If called from within a cancelled parent task, do not enqueue cron work.
+    throwIfAborted(parentRunContextManager.getContext()?.signal);
+
     const cronDef = this.client.crons.create(this.definition.name, {
       expression,
       input: input as JsonObject,