make 'cancel' in mock activity env more well-defined, make cancellation details field in activity context private, make activity cancellation tests less flaky

Author: THardy98

packages/activity/src/index.ts

@@ -301,7 +301,7 @@ export class Context {
   /**
    * Holder object for activity cancellation details
    */
-  public readonly cancellationDetails: ActivityCancellationDetailsHolder;
+  private readonly _cancellationDetails: ActivityCancellationDetailsHolder;
 
   /**
    * **Not** meant to instantiated by Activity code, used by the worker.
@@ -323,7 +323,7 @@ export class Context {
     this.heartbeatFn = heartbeat;
     this.log = log;
     this.metricMeter = metricMeter;
-    this.cancellationDetails = details;
+    this._cancellationDetails = details;
   }
 
   /**
@@ -363,6 +363,10 @@ export class Context {
     });
     return Promise.race([this.cancelled.finally(() => clearTimeout(handle)), timer]);
   };
+
+  public cancellationDetails(): ActivityCancellationDetails | undefined {
+    return this._cancellationDetails.details;
+  }
 }
 
 /**
@@ -447,7 +451,7 @@ export function cancelled(): Promise<never> {
  * Returns the cancellation details for this activity, if any.
  */
 export function cancellationDetails(): ActivityCancellationDetails | undefined {
-  return Context.current().cancellationDetails.details;
+  return Context.current().cancellationDetails();
 }
 
 /**

packages/common/src/activity-cancellation-details.ts

@@ -4,6 +4,15 @@ export interface ActivityCancellationDetailsHolder {
   details?: ActivityCancellationDetails;
 }
 
+export interface ActivityCancellationDetailsOptions {
+  notFound?: boolean;
+  cancelRequested?: boolean;
+  paused?: boolean;
+  timedOut?: boolean;
+  workerShutdown?: boolean;
+  reset?: boolean;
+}
+
 /**
  * Provides the reasons for the activity's cancellation. Cancellation details are set once and do not change once set.
  */
@@ -15,20 +24,13 @@ export class ActivityCancellationDetails {
   readonly workerShutdown: boolean;
   readonly reset: boolean;
 
-  private constructor(
-    notFound: boolean = false,
-    cancelRequested: boolean = false,
-    paused: boolean = false,
-    timedOut: boolean = false,
-    workerShutdown: boolean = false,
-    reset: boolean = false
-  ) {
-    this.notFound = notFound;
-    this.cancelRequested = cancelRequested;
-    this.paused = paused;
-    this.timedOut = timedOut;
-    this.workerShutdown = workerShutdown;
-    this.reset = reset;
+  public constructor(options: ActivityCancellationDetailsOptions = {}) {
+    this.notFound = options.notFound ?? false;
+    this.cancelRequested = options.cancelRequested ?? false;
+    this.paused = options.paused ?? false;
+    this.timedOut = options.timedOut ?? false;
+    this.workerShutdown = options.workerShutdown ?? false;
+    this.reset = options.reset ?? false;
   }
 
   static fromProto(
@@ -37,13 +39,13 @@ export class ActivityCancellationDetails {
     if (proto == null) {
       return new ActivityCancellationDetails();
     }
-    return new ActivityCancellationDetails(
-      proto.isNotFound ?? false,
-      proto.isCancelled ?? false,
-      proto.isPaused ?? false,
-      proto.isTimedOut ?? false,
-      proto.isWorkerShutdown ?? false,
-      proto.isReset ?? false
-    );
+    return new ActivityCancellationDetails({
+      notFound: proto.isNotFound ?? false,
+      cancelRequested: proto.isCancelled ?? false,
+      paused: proto.isPaused ?? false,
+      timedOut: proto.isTimedOut ?? false,
+      workerShutdown: proto.isWorkerShutdown ?? false,
+      reset: proto.isReset ?? false
+    });
   }
 }

packages/test/src/helpers-integration.ts

@@ -285,24 +285,6 @@ export function configurableHelpers<T>(
   };
 }
 
-export async function assertPendingActivityExistsEventually(
-  handle: WorkflowHandle<workflow.Workflow>,
-  activityId: string,
-  timeoutMs: number
-): Promise<temporal.api.workflow.v1.IPendingActivityInfo> {
-  let activityInfo: temporal.api.workflow.v1.IPendingActivityInfo | undefined;
-  try {
-    await waitUntil(async () => {
-      const desc = await handle.describe();
-      activityInfo = desc.raw.pendingActivities?.find((pa) => pa.activityId === activityId);
-      return activityInfo !== undefined;
-    }, timeoutMs);
-  } catch {
-    throw new Error(`Unable to find pending activity for activity ${activityId}`);
-  }
-  return activityInfo as temporal.api.workflow.v1.IPendingActivityInfo;
-}
-
 export async function setActivityPauseState(handle: WorkflowHandle, activityId: string, pause: boolean): Promise<void> {
   const desc = await handle.describe();
   const req = {
@@ -319,11 +301,13 @@ export async function setActivityPauseState(handle: WorkflowHandle, activityId:
     await handle.client.workflowService.unpauseActivity(req);
   }
   await waitUntil(async () => {
-    const info = await assertPendingActivityExistsEventually(handle, activityId, 10000);
+    const { raw } = await handle.describe();
+    const activityInfo = raw.pendingActivities?.find((act) => act.activityId === activityId);
+    if (!activityInfo) return false;
     if (pause) {
-      return info.paused ?? false;
+      return activityInfo.paused ?? false;
     }
-    return !info.paused;
+    return !activityInfo.paused;
   }, 10000);
 }
 

packages/test/src/test-mockactivityenv.ts

@@ -24,7 +24,7 @@ test('MockActivityEnvironment emits heartbeat events and can be cancelled', asyn
   const env = new MockActivityEnvironment();
   env.on('heartbeat', (d: unknown) => {
     if (d === 6) {
-      env.cancel('test');
+      env.cancel('CANCELLED');
     }
   });
   await t.throwsAsync(

packages/testing/src/mocking-activity-environment.ts

@@ -9,10 +9,11 @@ import {
   defaultPayloadConverter,
   MetricMeter,
   noopMetricMeter,
+  ActivityCancellationDetails,
 } from '@temporalio/common';
 import { LoggerWithComposedMetadata } from '@temporalio/common/lib/logger';
 import { ActivityInterceptorsFactory, DefaultLogger } from '@temporalio/worker';
-import { Activity } from '@temporalio/worker/lib/activity';
+import { Activity, CancelReason } from '@temporalio/worker/lib/activity';
 
 export interface MockActivityEnvironmentOptions {
   interceptors?: ActivityInterceptorsFactory[];
@@ -30,7 +31,10 @@ export interface MockActivityEnvironmentOptions {
  * will immediately be in a cancelled state.
  */
 export class MockActivityEnvironment extends events.EventEmitter {
-  public cancel: (reason?: any, details?: any) => void = () => undefined;
+  public cancel: (
+    reason?: CancelReason, 
+    details?: ActivityCancellationDetails
+  ) => void = () => undefined;
   public readonly context: activity.Context;
   private readonly activity: Activity;
 
@@ -52,7 +56,15 @@ export class MockActivityEnvironment extends events.EventEmitter {
       opts?.interceptors ?? []
     );
     this.context = this.activity.context;
-    this.cancel = this.activity.cancel;
+    this.cancel = (
+      reason?: CancelReason,
+      details?: ActivityCancellationDetails
+    ) => {
+      // Default to CANCELLED if nothing provided.
+      const r = reason ?? "CANCELLED";
+      const d = details ?? new ActivityCancellationDetails({ cancelRequested: true });
+      this.activity.cancel(r, d); 
+    }
   }
 
   /**

packages/worker/src/activity.ts

@@ -144,7 +144,7 @@ export class Activity {
         (error instanceof CancelledFailure || isAbortError(error)) &&
         this.context.cancellationSignal.aborted
       ) {
-        if (this.context.cancellationDetails.details?.paused) {
+        if (this.context.cancellationDetails()?.paused) {
           this.workerLogger.debug('Activity paused', { durationMs });
         } else {
           this.workerLogger.debug('Activity completed as cancelled', { durationMs });
@@ -187,7 +187,7 @@ export class Activity {
           // Either a CancelledFailure that we threw or AbortError from AbortController
           if (err instanceof CancelledFailure) {
             // If cancel due to activity pause, emit an application failure for the pause.
-            if (this.context.cancellationDetails.details?.paused) {
+            if (this.context.cancellationDetails()?.paused) {
               return {
                 failed: {
                   failure: await encodeErrorToFailure(