extstore/surface exstore errors in client

packages/client/src/activity-client.ts

@@ -15,6 +15,7 @@ import {
   convertDeploymentVersion,
   decodePriority,
   decompileRetryPolicy,
+  ExternalStorageError,
 } from '@temporalio/common';
 import type { Duration } from '@temporalio/common/lib/time';
 import { msOptionalToTs, msToNumber, optionalTsToDate, optionalTsToMs } from '@temporalio/common/lib/time';
@@ -482,7 +483,10 @@ export class ActivityClient extends AsyncCompletionClient implements TypedActivi
       }
       throw new ServiceError(fallbackMessage, { cause: err });
     }
-    throw new ServiceError('Unexpected error while making gRPC request');
+    if (err instanceof ExternalStorageError) {
+      throw new ServiceError('External storage failed', { cause: err });
+    }
+    throw new ServiceError('Unexpected error while making gRPC request', { cause: err as Error });
   }
 }
 

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

@@ -1,6 +1,6 @@
 import { status as grpcStatus } from '@grpc/grpc-js';
 import type { ActivitySerializationContext, StorageDriverTargetInfo } from '@temporalio/common';
-import { ensureTemporalFailure } from '@temporalio/common';
+import { ensureTemporalFailure, ExternalStorageError } from '@temporalio/common';
 import {
   encodeErrorToFailure,
   encodeToPayloadsWithContext,
@@ -163,8 +163,10 @@ export class AsyncCompletionClient extends BaseClient {
       }
 
       throw new ActivityCompletionError(err.details || err.message);
+    } else if (err instanceof ExternalStorageError) {
+      throw new ActivityCompletionError('External storage failed', { cause: err });
     }
-    throw new ActivityCompletionError('Unexpected failure');
+    throw new ActivityCompletionError('Unexpected failure', { cause: err });
   }
 
   /**

packages/client/src/errors.ts

@@ -88,10 +88,18 @@ export class ActivityNotFoundError extends Error {}
 
 /**
  * Thrown by {@link AsyncCompletionClient} when trying to complete or heartbeat
- * an Activity for any reason apart from {@link ActivityNotFoundError}.
+ * an Activity for any reason apart from {@link ActivityNotFoundError}. If an
+ * underlying error exists, it will be stored in the `cause` property.
  */
 @SymbolBasedInstanceOfError('ActivityCompletionError')
-export class ActivityCompletionError extends Error {}
+export class ActivityCompletionError extends Error {
+  public readonly cause?: unknown;
+
+  constructor(message: string, opts?: { cause?: unknown }) {
+    super(message);
+    this.cause = opts?.cause;
+  }
+}
 
 /**
  * Thrown by {@link AsyncCompletionClient.heartbeat} when the Workflow has

packages/client/src/nexus-client.ts

@@ -24,7 +24,7 @@ import {
 import { filterNullAndUndefined } from '@temporalio/common/lib/internal-workflow';
 import { msOptionalToTs, optionalTsToDate, optionalTsToMs } from '@temporalio/common/lib/time';
 import { temporal } from '@temporalio/proto';
-import type { LoadedDataConverter } from '@temporalio/common';
+import { ExternalStorageError, type LoadedDataConverter } from '@temporalio/common';
 import type { SearchAttributeType, TypedSearchAttributeValue } from '@temporalio/common/lib/search-attributes';
 import { decode } from '@temporalio/common/lib/encoding';
 import type { BaseClientOptions, LoadedWithDefaults, WithDefaults } from './base-client';
@@ -404,11 +404,11 @@ export class NexusClient extends BaseClient {
       userMetadata,
     };
     const externalStorage = this.dataConverter.externalStorage;
-    if (externalStorage) {
-      await visit(req, walkStartNexusOperationExecutionRequest, extstoreStoreOptions(externalStorage));
-    }
     let res: temporal.api.workflowservice.v1.IStartNexusOperationExecutionResponse;
     try {
+      if (externalStorage) {
+        await visit(req, walkStartNexusOperationExecutionRequest, extstoreStoreOptions(externalStorage));
+      }
       res = await this.connection.workflowService.startNexusOperationExecution(req);
     } catch (err: unknown) {
       this.rethrowGrpcError(err, 'Failed to start Nexus operation', input.id);
@@ -481,15 +481,14 @@ export class NexusClient extends BaseClient {
     };
     for (;;) {
       let res: temporal.api.workflowservice.v1.IPollNexusOperationExecutionResponse;
+      const externalStorage = this.dataConverter.externalStorage;
       try {
         res = await this.connection.workflowService.pollNexusOperationExecution(req);
+        await visit(res, walkPollNexusOperationExecutionResponse, extstoreInboundOptions(externalStorage));
       } catch (err: unknown) {
         this.rethrowGrpcError(err, 'Failed to poll Nexus operation result', input.operationId);
       }
 
-      const externalStorage = this.dataConverter.externalStorage;
-      await visit(res, walkPollNexusOperationExecutionResponse, extstoreInboundOptions(externalStorage));
-
       // The operation is closed if we have a result or failure
       if (res.result) {
         return await decodeFromPayloadsAtIndex(this.dataConverter, 0, [res.result]);
@@ -511,13 +510,13 @@ export class NexusClient extends BaseClient {
       runId: input.runId ?? '',
     };
     let res: temporal.api.workflowservice.v1.IDescribeNexusOperationExecutionResponse;
+    const externalStorage = this.dataConverter.externalStorage;
     try {
       res = await this.connection.workflowService.describeNexusOperationExecution(req);
+      await visit(res, walkDescribeNexusOperationExecutionResponse, extstoreInboundOptions(externalStorage));
     } catch (err: unknown) {
       this.rethrowGrpcError(err, 'Failed to describe Nexus operation', input.operationId);
     }
-    const externalStorage = this.dataConverter.externalStorage;
-    await visit(res, walkDescribeNexusOperationExecutionResponse, extstoreInboundOptions(externalStorage));
     if (!res.info) {
       throw new ServiceError('Received invalid Nexus operation description from server: missing info');
     }
@@ -560,18 +559,18 @@ export class NexusClient extends BaseClient {
     let nextPageToken: Uint8Array | undefined = undefined;
     for (;;) {
       let response: temporal.api.workflowservice.v1.IListNexusOperationExecutionsResponse;
+      const externalStorage = this.dataConverter.externalStorage;
       try {
         response = await this.connection.workflowService.listNexusOperationExecutions({
           namespace: this.options.namespace,
           query: input.query,
           pageSize: input.pageSize,
           nextPageToken,
         });
+        await visit(response, walkListNexusOperationExecutionsResponse, extstoreInboundOptions(externalStorage));
       } catch (err: unknown) {
         this.rethrowGrpcError(err, 'Failed to list Nexus operations', undefined);
       }
-      const externalStorage = this.dataConverter.externalStorage;
-      await visit(response, walkListNexusOperationExecutionsResponse, extstoreInboundOptions(externalStorage));
       for (const raw of response.operations ?? []) {
         yield nexusOperationListInfoFromProto(raw);
       }
@@ -607,6 +606,9 @@ export class NexusClient extends BaseClient {
       throw new ServiceError(fallbackMessage, { cause: err });
     }
 
+    if (err instanceof ExternalStorageError) {
+      throw new ServiceError('External storage failed', { cause: err });
+    }
     throw new ServiceError('Unexpected error while making gRPC request', { cause: err as Error });
   }
 }

packages/client/src/schedule-client.ts

@@ -1,6 +1,6 @@
 import { randomUUID } from 'node:crypto';
 import { status as grpcStatus } from '@grpc/grpc-js';
-import type { Workflow } from '@temporalio/common';
+import { ExternalStorageError, type Workflow } from '@temporalio/common';
 import {
   decodeSearchAttributes,
   decodeTypedSearchAttributes,
@@ -414,20 +414,19 @@ export class ScheduleClient extends BaseClient {
     let nextPageToken: Uint8Array | undefined = undefined;
     for (;;) {
       let response: temporal.api.workflowservice.v1.ListSchedulesResponse;
+      const externalStorage = this.dataConverter.externalStorage;
       try {
         response = await this.workflowService.listSchedules({
           nextPageToken,
           namespace: this.options.namespace,
           maximumPageSize: options?.pageSize,
           query: options?.query,
         });
+        await visit(response, walkListSchedulesResponse, extstoreInboundOptions(externalStorage));
       } catch (e) {
         this.rethrowGrpcError(e, 'Failed to list schedules', undefined);
       }
 
-      const externalStorage = this.dataConverter.externalStorage;
-      await visit(response, walkListSchedulesResponse, extstoreInboundOptions(externalStorage));
-
       for (const raw of response.schedules ?? []) {
         yield <ScheduleSummary>{
           scheduleId: raw.scheduleId,
@@ -577,6 +576,9 @@ export class ScheduleClient extends BaseClient {
 
       throw new ServiceError(fallbackMessage, { cause: err });
     }
+    if (err instanceof ExternalStorageError) {
+      throw new ServiceError('External storage failed', { cause: err });
+    }
     throw new ServiceError('Unexpected error while making gRPC request', { cause: err as Error });
   }
 }

packages/client/src/workflow-client.ts

@@ -26,6 +26,7 @@ import {
   decodeRetryState,
   encodeWorkflowIdConflictPolicy,
   compilePriority,
+  ExternalStorageError,
 } from '@temporalio/common';
 import { encodeUserMetadata } from '@temporalio/common/lib/internal-non-workflow/codec-helpers';
 import { encodeUnifiedSearchAttributes } from '@temporalio/common/lib/converter/payload-search-attributes';
@@ -830,13 +831,13 @@ export class WorkflowClient extends BaseClient {
 
     for (;;) {
       let res: temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse;
+      const externalStorage = this.dataConverter.externalStorage;
       try {
         res = await this.workflowService.getWorkflowExecutionHistory(req);
+        await visit(res, walkGetWorkflowExecutionHistoryResponse, extstoreInboundOptions(externalStorage));
       } catch (err) {
         this.rethrowGrpcError(err, 'Failed to get Workflow execution history', { workflowId, runId });
       }
-      const externalStorage = this.dataConverter.externalStorage;
-      await visit(res, walkGetWorkflowExecutionHistoryResponse, extstoreInboundOptions(externalStorage));
       const events = res.history?.events;
 
       if (events == null || events.length === 0) {
@@ -960,6 +961,9 @@ export class WorkflowClient extends BaseClient {
 
       throw new ServiceError(fallbackMessage, { cause: err });
     }
+    if (err instanceof ExternalStorageError) {
+      throw new ServiceError('External storage failed', { cause: err });
+    }
     throw new ServiceError('Unexpected error while making gRPC request', { cause: err as Error });
   }
 
@@ -982,22 +986,23 @@ export class WorkflowClient extends BaseClient {
       },
     };
     const externalStorage = this.dataConverter.externalStorage;
-    if (externalStorage) {
-      await visit(
-        req,
-        walkQueryWorkflowRequest,
-        extstoreStoreOptions(externalStorage, {
-          initialTarget: {
-            kind: 'workflow',
-            namespace: this.options.namespace,
-            id: input.workflowExecution.workflowId ?? undefined,
-          },
-        })
-      );
-    }
     let response: temporal.api.workflowservice.v1.QueryWorkflowResponse;
     try {
+      if (externalStorage) {
+        await visit(
+          req,
+          walkQueryWorkflowRequest,
+          extstoreStoreOptions(externalStorage, {
+            initialTarget: {
+              kind: 'workflow',
+              namespace: this.options.namespace,
+              id: input.workflowExecution.workflowId ?? undefined,
+            },
+          })
+        );
+      }
       response = await this.workflowService.queryWorkflow(req);
+      await visit(response, walkQueryWorkflowResponse, extstoreInboundOptions(externalStorage));
     } catch (err) {
       if (isGrpcServiceError(err)) {
         rethrowKnownErrorTypes(err);
@@ -1007,7 +1012,6 @@ export class WorkflowClient extends BaseClient {
       }
       this.rethrowGrpcError(err, 'Failed to query Workflow', input.workflowExecution);
     }
-    await visit(response, walkQueryWorkflowResponse, extstoreInboundOptions(externalStorage));
     if (response.queryRejected) {
       if (response.queryRejected.status === undefined || response.queryRejected.status === null) {
         throw new TypeError('Received queryRejected from server with no status');
@@ -1069,34 +1073,34 @@ export class WorkflowClient extends BaseClient {
 
     const request = await this._createUpdateWorkflowRequest(waitForStageProto, input);
     const externalStorage = this.dataConverter.externalStorage;
-    if (externalStorage) {
-      await visit(
-        request,
-        walkUpdateWorkflowExecutionRequest,
-        extstoreStoreOptions(externalStorage, {
-          initialTarget: {
-            kind: 'workflow',
-            namespace: this.options.namespace,
-            id: input.workflowExecution.workflowId ?? undefined,
-          },
-        })
-      );
-    }
 
     // Repeatedly send UpdateWorkflowExecution until update is durable (if the server receives a request with
     // an update ID that already exists, it responds with information for the existing update). If the
     // requested wait stage is COMPLETED, further polling is done before returning the UpdateHandle.
     let response: temporal.api.workflowservice.v1.UpdateWorkflowExecutionResponse;
     try {
+      if (externalStorage) {
+        await visit(
+          request,
+          walkUpdateWorkflowExecutionRequest,
+          extstoreStoreOptions(externalStorage, {
+            initialTarget: {
+              kind: 'workflow',
+              namespace: this.options.namespace,
+              id: input.workflowExecution.workflowId ?? undefined,
+            },
+          })
+        );
+      }
       do {
         response = await this.workflowService.updateWorkflowExecution(request);
       } while (
         response.stage < UpdateWorkflowExecutionLifecycleStage.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_ACCEPTED
       );
+      await visit(response, walkUpdateWorkflowExecutionResponse, extstoreInboundOptions(externalStorage));
     } catch (err) {
       this.rethrowUpdateGrpcError(err, 'Workflow Update failed', input.workflowExecution);
     }
-    await visit(response, walkUpdateWorkflowExecutionResponse, extstoreInboundOptions(externalStorage));
     return {
       updateId: request.request!.meta!.updateId!,
 
@@ -1759,18 +1763,18 @@ export class WorkflowClient extends BaseClient {
     let nextPageToken: Uint8Array = Buffer.alloc(0);
     for (;;) {
       let response: temporal.api.workflowservice.v1.ListWorkflowExecutionsResponse;
+      const externalStorage = this.dataConverter.externalStorage;
       try {
         response = await this.workflowService.listWorkflowExecutions({
           namespace: this.options.namespace,
           query: options?.query,
           nextPageToken,
           pageSize: options?.pageSize,
         });
+        await visit(response, walkListWorkflowExecutionsResponse, extstoreInboundOptions(externalStorage));
       } catch (e) {
         this.rethrowGrpcError(e, 'Failed to list workflows', undefined);
       }
-      const externalStorage = this.dataConverter.externalStorage;
-      await visit(response, walkListWorkflowExecutionsResponse, extstoreInboundOptions(externalStorage));
       // Not decoding memo payloads concurrently even though we could have to keep the lazy nature of this iterator.
       // Decoding is done for `memo` fields which tend to be small.
       // We might decide to change that based on user feedback.

packages/common/src/errors.ts

@@ -75,14 +75,55 @@ export class NamespaceNotFoundError extends Error {
 export class CompleteAsyncError extends Error {}
 
 /**
- * Thrown when an inbound payload is detected as an external-storage reference
+ * Base type for all external storage errors.
+ *
+ * @experimental
+ */
+@SymbolBasedInstanceOfError('ExternalStorageError')
+export abstract class ExternalStorageError extends Error {}
+
+/**
+ * [TMPRL1105] Thrown when an inbound payload is detected as an external-storage reference
  * but no `ExternalStorage` is configured to resolve it.
  *
  * @experimental
  */
 @SymbolBasedInstanceOfError('ExternalStorageNotConfiguredError')
-export class ExternalStorageNotConfiguredError extends Error {
+export class ExternalStorageNotConfiguredError extends ExternalStorageError {
   constructor(message = 'Detected externally stored payload(s) but external storage is not configured.') {
     super(`[TMPRL1105] ${message}`);
   }
 }
+
+/**
+ * Thrown when a storage driver's `store` or `retrieve` operation fails.
+ *
+ * @experimental
+ */
+@SymbolBasedInstanceOfError('ExternalStorageDriverError')
+export class ExternalStorageDriverError extends ExternalStorageError {
+  public readonly cause?: unknown;
+
+  constructor(message: string, opts?: { cause?: unknown }) {
+    super(message);
+    this.cause = opts?.cause;
+  }
+}
+
+/**
+ * Thrown when external storage is asked to use a driver that is not registered
+ * on the configured `ExternalStorage`.
+ *
+ * @experimental
+ */
+@SymbolBasedInstanceOfError('ExternalStorageUnregisteredDriverError')
+export class ExternalStorageUnregisteredDriverError extends ExternalStorageError {}
+
+/**
+ * Thrown when an external-storage reference is malformed, or a driver returns a result inconsistent
+ * with its input.
+ *
+ * @experimental
+ */
+@SymbolBasedInstanceOfError('ExternalStorageReferenceError')
+export class ExternalStorageReferenceError extends ExternalStorageError {}