feat(client): Throw more specific errors from Client APIs (#1147)

## What changed

- All Client APIs now throw `NamespaceNotFoundError` when appropriate,
rather than `WorkflowNotFound`, `ScheduleNotFound` or `ActivityNotFound`
(fix #1145, fix #976)
- Errors that happen while preparing API requests from a `Client`, and
particularly errors happening while converting input objects to
payloads, now results in throwing an appropriate exception, rather than
an opaque ServiceError.

---------

Co-authored-by: Roey Berman <roey.berman@gmail.com>

Author: mjameswh

packages/client/src/async-completion-client.ts

@@ -1,5 +1,6 @@
-import { Status } from '@grpc/grpc-js/build/src/constants';
+import { status as grpcStatus } from '@grpc/grpc-js';
 import { ensureTemporalFailure } from '@temporalio/common';
+import type { temporal } from '@temporalio/proto';
 import {
   encodeErrorToFailure,
   encodeToPayloads,
@@ -12,8 +13,9 @@ import {
   LoadedWithDefaults,
   WithDefaults,
 } from './base-client';
-import { isServerErrorResponse } from './errors';
+import { isGrpcServiceError } from './errors';
 import { WorkflowService } from './types';
+import { rethrowKnownErrorTypes } from './helpers';
 
 /**
  * Thrown by {@link AsyncCompletionClient} when trying to complete or heartbeat an Activity that does not exist in the
@@ -95,10 +97,13 @@ export class AsyncCompletionClient extends BaseClient {
    * Transforms grpc errors into well defined TS errors.
    */
   protected handleError(err: unknown): never {
-    if (isServerErrorResponse(err)) {
-      if (err.code === Status.NOT_FOUND) {
+    if (isGrpcServiceError(err)) {
+      rethrowKnownErrorTypes(err);
+
+      if (err.code === grpcStatus.NOT_FOUND) {
         throw new ActivityNotFoundError('Not found');
       }
+
       throw new ActivityCompletionError(err.details || err.message);
     }
     throw new ActivityCompletionError('Unexpected failure');
@@ -114,20 +119,21 @@ export class AsyncCompletionClient extends BaseClient {
   async complete(fullActivityId: FullActivityId, result: unknown): Promise<void>;
 
   async complete(taskTokenOrFullActivityId: Uint8Array | FullActivityId, result: unknown): Promise<void> {
+    const payloads = await encodeToPayloads(this.dataConverter, result);
     try {
       if (taskTokenOrFullActivityId instanceof Uint8Array) {
         await this.workflowService.respondActivityTaskCompleted({
           identity: this.options.identity,
           namespace: this.options.namespace,
           taskToken: taskTokenOrFullActivityId,
-          result: { payloads: await encodeToPayloads(this.dataConverter, result) },
+          result: { payloads },
         });
       } else {
         await this.workflowService.respondActivityTaskCompletedById({
           identity: this.options.identity,
           namespace: this.options.namespace,
           ...taskTokenOrFullActivityId,
-          result: { payloads: await encodeToPayloads(this.dataConverter, result) },
+          result: { payloads },
         });
       }
     } catch (err) {
@@ -145,20 +151,21 @@ export class AsyncCompletionClient extends BaseClient {
   async fail(fullActivityId: FullActivityId, err: unknown): Promise<void>;
 
   async fail(taskTokenOrFullActivityId: Uint8Array | FullActivityId, err: unknown): Promise<void> {
+    const failure = await encodeErrorToFailure(this.dataConverter, ensureTemporalFailure(err));
     try {
       if (taskTokenOrFullActivityId instanceof Uint8Array) {
         await this.workflowService.respondActivityTaskFailed({
           identity: this.options.identity,
           namespace: this.options.namespace,
           taskToken: taskTokenOrFullActivityId,
-          failure: await encodeErrorToFailure(this.dataConverter, ensureTemporalFailure(err)),
+          failure,
         });
       } else {
         await this.workflowService.respondActivityTaskFailedById({
           identity: this.options.identity,
           namespace: this.options.namespace,
           ...taskTokenOrFullActivityId,
-          failure: await encodeErrorToFailure(this.dataConverter, err),
+          failure,
         });
       }
     } catch (err) {
@@ -176,20 +183,21 @@ export class AsyncCompletionClient extends BaseClient {
   reportCancellation(fullActivityId: FullActivityId, details?: unknown): Promise<void>;
 
   async reportCancellation(taskTokenOrFullActivityId: Uint8Array | FullActivityId, details?: unknown): Promise<void> {
+    const payloads = await encodeToPayloads(this.dataConverter, details);
     try {
       if (taskTokenOrFullActivityId instanceof Uint8Array) {
         await this.workflowService.respondActivityTaskCanceled({
           identity: this.options.identity,
           namespace: this.options.namespace,
           taskToken: taskTokenOrFullActivityId,
-          details: { payloads: await encodeToPayloads(this.dataConverter, details) },
+          details: { payloads },
         });
       } else {
         await this.workflowService.respondActivityTaskCanceledById({
           identity: this.options.identity,
           namespace: this.options.namespace,
           ...taskTokenOrFullActivityId,
-          details: { payloads: await encodeToPayloads(this.dataConverter, details) },
+          details: { payloads },
         });
       }
     } catch (err) {
@@ -207,36 +215,31 @@ export class AsyncCompletionClient extends BaseClient {
   heartbeat(fullActivityId: FullActivityId, details?: unknown): Promise<void>;
 
   async heartbeat(taskTokenOrFullActivityId: Uint8Array | FullActivityId, details?: unknown): Promise<void> {
+    const payloads = await encodeToPayloads(this.dataConverter, details);
     let cancelRequested = false;
     try {
-      const response = await this._sendHeartbeat(taskTokenOrFullActivityId, details);
-      cancelRequested = response.cancelRequested;
+      let response: temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse;
+      if (taskTokenOrFullActivityId instanceof Uint8Array) {
+        response = await this.workflowService.recordActivityTaskHeartbeat({
+          identity: this.options.identity,
+          namespace: this.options.namespace,
+          taskToken: taskTokenOrFullActivityId,
+          details: { payloads },
+        });
+      } else {
+        response = await this.workflowService.recordActivityTaskHeartbeatById({
+          identity: this.options.identity,
+          namespace: this.options.namespace,
+          ...taskTokenOrFullActivityId,
+          details: { payloads },
+        });
+      }
+      cancelRequested = !!response.cancelRequested;
     } catch (err) {
       this.handleError(err);
     }
     if (cancelRequested) {
       throw new ActivityCancelledError('cancelled');
     }
   }
-
-  private async _sendHeartbeat(
-    taskTokenOrFullActivityId: Uint8Array | FullActivityId,
-    details?: unknown
-  ): Promise<{ cancelRequested: boolean }> {
-    if (taskTokenOrFullActivityId instanceof Uint8Array) {
-      return await this.workflowService.recordActivityTaskHeartbeat({
-        identity: this.options.identity,
-        namespace: this.options.namespace,
-        taskToken: taskTokenOrFullActivityId,
-        details: { payloads: await encodeToPayloads(this.dataConverter, details) },
-      });
-    } else {
-      return await this.workflowService.recordActivityTaskHeartbeatById({
-        identity: this.options.identity,
-        namespace: this.options.namespace,
-        ...taskTokenOrFullActivityId,
-        details: { payloads: await encodeToPayloads(this.dataConverter, details) },
-      });
-    }
-  }
 }

packages/client/src/connection.ts

@@ -3,7 +3,7 @@ import * as grpc from '@grpc/grpc-js';
 import type { RPCImpl } from 'protobufjs';
 import { filterNullAndUndefined, normalizeTlsConfig, TLSConfig } from '@temporalio/common/lib/internal-non-workflow';
 import { Duration, msOptionalToNumber } from '@temporalio/common/lib/time';
-import { isServerErrorResponse, ServiceError } from './errors';
+import { isGrpcServiceError, ServiceError } from './errors';
 import { defaultGrpcRetryOptions, makeGrpcRetryInterceptor } from './grpc-retry';
 import pkg from './pkg';
 import { CallContext, HealthService, Metadata, OperatorService, WorkflowService } from './types';
@@ -276,7 +276,7 @@ export class Connection {
         try {
           await this.withDeadline(deadline, () => this.workflowService.getSystemInfo({}));
         } catch (err) {
-          if (isServerErrorResponse(err)) {
+          if (isGrpcServiceError(err)) {
             // Ignore old servers
             if (err.code !== grpc.status.UNIMPLEMENTED) {
               throw new ServiceError('Failed to connect to Temporal server', { cause: err });

packages/client/src/errors.ts

@@ -1,4 +1,4 @@
-import { ServerErrorResponse } from '@grpc/grpc-js';
+import { ServiceError as GrpcServiceError } from '@grpc/grpc-js';
 import { RetryState, TemporalFailure } from '@temporalio/common';
 
 /**
@@ -47,10 +47,11 @@ export class WorkflowContinuedAsNewError extends Error {
   }
 }
 
+export function isGrpcServiceError(err: unknown): err is GrpcServiceError {
+  return err instanceof Error && (err as any).details !== undefined && (err as any).metadata !== undefined;
+}
+
 /**
- * Type assertion helper, assertion is mostly empty because any additional
- * properties are optional.
+ * @deprecated Use `isGrpcServiceError` instead
  */
-export function isServerErrorResponse(err: unknown): err is ServerErrorResponse {
-  return err instanceof Error;
-}
+export const isServerErrorResponse = isGrpcServiceError;

packages/client/src/helpers.ts

@@ -1,13 +1,15 @@
+import { ServiceError as GrpcServiceError } from '@grpc/grpc-js';
 import {
   LoadedDataConverter,
   mapFromPayloads,
+  NamespaceNotFoundError,
   searchAttributePayloadConverter,
   SearchAttributes,
 } from '@temporalio/common';
 import { Replace } from '@temporalio/common/lib/type-helpers';
 import { optionalTsToDate, tsToDate } from '@temporalio/common/lib/time';
 import { decodeMapFromPayloads } from '@temporalio/common/lib/internal-non-workflow/codec-helpers';
-import { temporal } from '@temporalio/proto';
+import { temporal, google } from '@temporalio/proto';
 import { RawWorkflowExecutionInfo, WorkflowExecutionInfo, WorkflowExecutionStatusName } from './types';
 
 function workflowStatusCodeToName(code: temporal.api.enums.v1.WorkflowExecutionStatus): WorkflowExecutionStatusName {
@@ -75,3 +77,36 @@ export async function executionInfoFromRaw<T>(
     raw: rawDataToEmbed,
   };
 }
+
+type ErrorDetailsName = `temporal.api.errordetails.v1.${keyof typeof temporal.api.errordetails.v1}`;
+
+/**
+ * If the error type can be determined based on embedded grpc error details,
+ * then rethrow the appropriate TypeScript error. Otherwise do nothing.
+ *
+ * This function should be used before falling back to generic error handling
+ * based on grpc error code. Very few error types are currently supported, but
+ * this function will be expanded over time as more server error types are added.
+ */
+export function rethrowKnownErrorTypes(err: GrpcServiceError): void {
+  // We really don't expect multiple error details, but this really is an array, so just in case...
+  for (const entry of getGrpcStatusDetails(err) ?? []) {
+    if (!entry.type_url || !entry.value) continue;
+    const type = entry.type_url.replace(/^type.googleapis.com\//, '') as ErrorDetailsName;
+
+    switch (type) {
+      case 'temporal.api.errordetails.v1.NamespaceNotFoundFailure': {
+        const { namespace } = temporal.api.errordetails.v1.NamespaceNotFoundFailure.decode(entry.value);
+        throw new NamespaceNotFoundError(namespace);
+      }
+    }
+  }
+}
+
+function getGrpcStatusDetails(err: GrpcServiceError): google.rpc.Status['details'] | undefined {
+  const statusBuffer = err.metadata.get('grpc-status-details-bin')?.[0];
+  if (!statusBuffer || typeof statusBuffer === 'string') {
+    return undefined;
+  }
+  return google.rpc.Status.decode(statusBuffer).details;
+}

packages/client/src/schedule-client.ts

@@ -11,7 +11,7 @@ import { temporal } from '@temporalio/proto';
 import { optionalDateToTs, optionalTsToDate, optionalTsToMs, tsToDate } from '@temporalio/common/lib/time';
 import { CreateScheduleInput, CreateScheduleOutput, ScheduleClientInterceptor } from './interceptors';
 import { WorkflowService } from './types';
-import { isServerErrorResponse, ServiceError } from './errors';
+import { isGrpcServiceError, ServiceError } from './errors';
 import {
   Backfill,
   CompiledScheduleUpdateOptions,
@@ -45,6 +45,7 @@ import {
   LoadedWithDefaults,
   WithDefaults,
 } from './base-client';
+import { rethrowKnownErrorTypes } from './helpers';
 
 /**
  * Handle to a single Schedule
@@ -263,7 +264,7 @@ export class ScheduleClient extends BaseClient {
       if (err.code === grpcStatus.ALREADY_EXISTS) {
         throw new ScheduleAlreadyRunning('Schedule already exists and is running', opts.scheduleId);
       }
-      this.rethrowGrpcError(err, opts.scheduleId, 'Failed to create schedule');
+      this.rethrowGrpcError(err, 'Failed to create schedule', opts.scheduleId);
     }
   }
 
@@ -279,7 +280,7 @@ export class ScheduleClient extends BaseClient {
         scheduleId,
       });
     } catch (err: any) {
-      this.rethrowGrpcError(err, scheduleId, 'Failed to describe schedule');
+      this.rethrowGrpcError(err, 'Failed to describe schedule', scheduleId);
     }
   }
 
@@ -306,7 +307,7 @@ export class ScheduleClient extends BaseClient {
     try {
       return await this.workflowService.updateSchedule(req);
     } catch (err: any) {
-      this.rethrowGrpcError(err, scheduleId, 'Failed to update schedule');
+      this.rethrowGrpcError(err, 'Failed to update schedule', scheduleId);
     }
   }
 
@@ -326,7 +327,7 @@ export class ScheduleClient extends BaseClient {
         patch,
       });
     } catch (err: any) {
-      this.rethrowGrpcError(err, scheduleId, 'Failed to patch schedule');
+      this.rethrowGrpcError(err, 'Failed to patch schedule', scheduleId);
     }
   }
 
@@ -343,7 +344,7 @@ export class ScheduleClient extends BaseClient {
         scheduleId,
       });
     } catch (err: any) {
-      this.rethrowGrpcError(err, scheduleId, 'Failed to delete schedule');
+      this.rethrowGrpcError(err, 'Failed to delete schedule', scheduleId);
     }
   }
 
@@ -366,13 +367,16 @@ export class ScheduleClient extends BaseClient {
   public async *list(options?: ListScheduleOptions): AsyncIterable<ScheduleSummary> {
     let nextPageToken: Uint8Array | undefined = undefined;
     for (;;) {
-      const response: temporal.api.workflowservice.v1.IListSchedulesResponse = await this.workflowService.listSchedules(
-        {
+      let response: temporal.api.workflowservice.v1.ListSchedulesResponse;
+      try {
+        response = await this.workflowService.listSchedules({
           nextPageToken,
           namespace: this.options.namespace,
           maximumPageSize: options?.pageSize,
-        }
-      );
+        });
+      } catch (e) {
+        this.rethrowGrpcError(e, 'Failed to list schedules', undefined);
+      }
 
       for (const raw of response.schedules ?? []) {
         yield <ScheduleSummary>{
@@ -497,20 +501,23 @@ export class ScheduleClient extends BaseClient {
     };
   }
 
-  protected rethrowGrpcError(err: unknown, scheduleId: string, fallbackMessage: string): never {
-    if (isServerErrorResponse(err)) {
+  protected rethrowGrpcError(err: unknown, fallbackMessage: string, scheduleId?: string): never {
+    if (isGrpcServiceError(err)) {
+      rethrowKnownErrorTypes(err);
+
       if (err.code === grpcStatus.NOT_FOUND) {
-        throw new ScheduleNotFoundError(err.details ?? 'Schedule not found', scheduleId);
+        throw new ScheduleNotFoundError(err.details ?? 'Schedule not found', scheduleId ?? '');
       }
       if (
         err.code === grpcStatus.INVALID_ARGUMENT &&
         err.message.match(/^3 INVALID_ARGUMENT: Invalid schedule spec: /)
       ) {
         throw new TypeError(err.message.replace(/^3 INVALID_ARGUMENT: Invalid schedule spec: /, ''));
       }
+
       throw new ServiceError(fallbackMessage, { cause: err });
     }
-    throw new ServiceError('Unexpected error while making gRPC request');
+    throw new ServiceError('Unexpected error while making gRPC request', { cause: err as Error });
   }
 }