feat: rework error handler

packages/datastore/src/datastore/datastore.ts

@@ -565,7 +565,7 @@ function defaultConflictHandler(conflictData: SyncConflict): PersistentModel {
 	return modelInstanceCreator(modelConstructor, { ...localModel, _version });
 }
 
-function defaultErrorHandler(error: SyncError) {
+function defaultErrorHandler(error: SyncError<PersistentModel>): void {
 	logger.warn(error);
 }
 
@@ -686,7 +686,7 @@ class DataStore {
 	private amplifyConfig: Record<string, any> = {};
 	private authModeStrategy: AuthModeStrategy;
 	private conflictHandler: ConflictHandler;
-	private errorHandler: (error: SyncError) => void;
+	private errorHandler: (error: SyncError<PersistentModel>) => void;
 	private fullSyncInterval: number;
 	private initialized: Promise<void>;
 	private initReject: Function;

packages/datastore/src/sync/index.ts

@@ -136,13 +136,15 @@ export class SyncEngine {
 			this.schema,
 			this.syncPredicates,
 			this.amplifyConfig,
-			this.authModeStrategy
+			this.authModeStrategy,
+			errorHandler
 		);
 		this.subscriptionsProcessor = new SubscriptionProcessor(
 			this.schema,
 			this.syncPredicates,
 			this.amplifyConfig,
-			this.authModeStrategy
+			this.authModeStrategy,
+			errorHandler
 		);
 		this.mutationsProcessor = new MutationProcessor(
 			this.schema,

packages/datastore/src/sync/processors/mutation.ts

@@ -24,6 +24,7 @@ import {
 	PersistentModelConstructor,
 	SchemaModel,
 	TypeConstructorMap,
+	ProcessName,
 } from '../../types';
 import { exhaustiveCheck, USER } from '../../util';
 import { MutationEventOutbox } from '../outbox';
@@ -33,12 +34,18 @@ import {
 	getModelAuthModes,
 	TransformerMutationType,
 	getTokenForCustomAuth,
+	ErrorMap,
+	mapErrorToType,
 } from '../utils';
 
 const MAX_ATTEMPTS = 10;
 
 const logger = new Logger('DataStore');
 
+const errorMap = {
+	BadRecord: error => /^Cannot return \w+ for [\w-_]+ type/.test(error.message),
+} as ErrorMap;
+
 type MutationProcessorEvent = {
 	operation: TransformerMutationType;
 	modelDefinition: SchemaModel;
@@ -373,20 +380,21 @@ class MutationProcessor {
 							} else {
 								try {
 									await this.errorHandler({
-										localModel: this.modelInstanceCreator(
-											modelConstructor,
-											variables.input
-										),
+										recoverySuggestion:
+											'Ensure app code is up to date, auth directives exist and are correct on each model, and that server-side data has not been invalidated by a schema change. If the problem persists, search for or create an issue: https://github.com/aws-amplify/amplify-js/issues',
+										localModel: variables.input,
 										message: error.message,
 										operation,
-										errorType: error.errorType,
+										errorType: mapErrorToType(errorMap, error),
 										errorInfo: error.errorInfo,
+										process: ProcessName.mutate,
+										cause: error,
 										remoteModel: error.data
 											? this.modelInstanceCreator(modelConstructor, error.data)
 											: null,
 									});
 								} catch (err) {
-									logger.warn('failed to execute errorHandler', err);
+									logger.warn('Mutation error handler failed with:', err);
 								} finally {
 									// Return empty tuple, dequeues the mutation
 									return error.data

packages/datastore/src/sync/processors/subscription.ts

@@ -12,6 +12,8 @@ import {
 	PredicatesGroup,
 	ModelPredicate,
 	AuthModeStrategy,
+	ErrorHandler,
+	ProcessName,
 } from '../../types';
 import {
 	buildSubscriptionGraphQLOperation,
@@ -20,12 +22,26 @@ import {
 	getUserGroupsFromToken,
 	TransformerMutationType,
 	getTokenForCustomAuth,
+	mapErrorToType,
+	ErrorMap,
 } from '../utils';
 import { ModelPredicateCreator } from '../../predicates';
 import { validatePredicate } from '../../util';
 
 const logger = new Logger('DataStore');
 
+const errorMap = {
+	Unauthorized: (givenError: any) => {
+		const {
+			error: { errors: [{ message = '' } = {}] } = {
+				errors: [],
+			},
+		} = givenError;
+		const regex = /Connection failed.+Unauthorized/;
+		return regex.test(message);
+	},
+} as ErrorMap;
+
 export enum CONTROL_MSG {
 	CONNECTED = 'CONNECTED',
 }
@@ -56,7 +72,8 @@ class SubscriptionProcessor {
 		private readonly schema: InternalSchema,
 		private readonly syncPredicates: WeakMap<SchemaModel, ModelPredicate<any>>,
 		private readonly amplifyConfig: Record<string, any> = {},
-		private readonly authModeStrategy: AuthModeStrategy
+		private readonly authModeStrategy: AuthModeStrategy,
+		private readonly errorHandler?: ErrorHandler
 	) {}
 
 	private buildSubscription(
@@ -436,12 +453,31 @@ class SubscriptionProcessor {
 												}
 												this.drainBuffer();
 											},
-											error: subscriptionError => {
+											error: async subscriptionError => {
 												const {
 													error: { errors: [{ message = '' } = {}] } = {
 														errors: [],
 													},
 												} = subscriptionError;
+												try {
+													await this.errorHandler({
+														recoverySuggestion:
+															'Ensure app code is up to date, auth directives exist and are correct on each model, and that server-side data has not been invalidated by a schema change. If the problem persists, search for or create an issue: https://github.com/aws-amplify/amplify-js/issues',
+														localModel: null,
+														message: message,
+														model: modelDefinition.name,
+														operation: operation,
+														errorType: mapErrorToType(
+															errorMap,
+															subscriptionError
+														),
+														process: ProcessName.subscribe,
+														remoteModel: null,
+														cause: subscriptionError,
+													});
+												} catch (e) {
+													logger.error('Sync error handler failed with:', e);
+												}
 
 												if (
 													message.includes(
@@ -487,7 +523,6 @@ class SubscriptionProcessor {
 														return;
 													}
 												}
-
 												logger.warn('subscriptionError', message);
 
 												if (typeof subscriptionReadyCallback === 'function') {
@@ -500,7 +535,6 @@ class SubscriptionProcessor {
 												) {
 													return;
 												}
-
 												observer.error(message);
 											},
 										})

packages/datastore/src/sync/processors/sync.ts

@@ -8,6 +8,8 @@ import {
 	PredicatesGroup,
 	GraphQLFilter,
 	AuthModeStrategy,
+	ErrorHandler,
+	ProcessName,
 } from '../../types';
 import {
 	buildGraphQLOperation,
@@ -16,6 +18,8 @@ import {
 	getForbiddenError,
 	predicateToGraphQLFilter,
 	getTokenForCustomAuth,
+	mapErrorToType,
+	ErrorMap,
 } from '../utils';
 import {
 	jitteredExponentialRetry,
@@ -24,7 +28,7 @@ import {
 	NonRetryableError,
 } from '@aws-amplify/core';
 import { ModelPredicateCreator } from '../../predicates';
-
+import { ModelInstanceCreator } from '../../datastore/datastore';
 const opResultDefaults = {
 	items: [],
 	nextToken: null,
@@ -33,14 +37,20 @@ const opResultDefaults = {
 
 const logger = new Logger('DataStore');
 
+const errorMap = {
+	BadRecord: error => /^Cannot return \w+ for [\w-_]+ type/.test(error.message),
+} as ErrorMap;
+
 class SyncProcessor {
 	private readonly typeQuery = new WeakMap<SchemaModel, [string, string]>();
 
 	constructor(
 		private readonly schema: InternalSchema,
 		private readonly syncPredicates: WeakMap<SchemaModel, ModelPredicate<any>>,
 		private readonly amplifyConfig: Record<string, any> = {},
-		private readonly authModeStrategy: AuthModeStrategy
+		private readonly authModeStrategy: AuthModeStrategy,
+		private readonly errorHandler?: ErrorHandler,
+		private readonly modelInstanceCreator?: ModelInstanceCreator
 	) {
 		this.generateQueries();
 	}
@@ -223,15 +233,33 @@ class SyncProcessor {
 							error.data[opName] &&
 							error.data[opName].items
 					);
-
 					if (this.partialDataFeatureFlagEnabled()) {
 						if (hasItems) {
 							const result = error;
 							result.data[opName].items = result.data[opName].items.filter(
 								item => item !== null
 							);
-
 							if (error.errors) {
+								await Promise.all(
+									error.error.map(async err => {
+										try {
+											await this.errorHandler({
+												recoverySuggestion:
+													'Ensure app code is up to date, auth directives exist and are correct on each model, and that server-side data has not been invalidated by a schema change. If the problem persists, search for or create an issue: https://github.com/aws-amplify/amplify-js/issues',
+												localModel: null,
+												message: err.message,
+												model: modelDefinition.name,
+												operation: opName,
+												errorType: mapErrorToType(errorMap, err),
+												process: ProcessName.sync,
+												remoteModel: null,
+												cause: err,
+											});
+										} catch (e) {
+											logger.error('Sync error handler failed with:', e);
+										}
+									})
+								);
 								Hub.dispatch('datastore', {
 									event: 'syncQueriesPartialSyncError',
 									data: {

packages/datastore/src/sync/utils.ts

@@ -26,6 +26,7 @@ import {
 	ModelOperation,
 	InternalSchema,
 	AuthModeStrategy,
+	ErrorType,
 } from '../types';
 import { exhaustiveCheck } from '../util';
 import { MutationEvent } from './';
@@ -40,6 +41,27 @@ enum GraphQLOperationType {
 	GET = 'query',
 }
 
+export type ErrorMap = {
+	[key in ErrorType]: (error: Error) => boolean;
+};
+
+/**
+ * Categorizes an error with a broad error type, intended to make
+ * customer error handling code simpler.
+ * @param errorMap Error names and a list of patterns that indicate them (each pattern as a regex or function)
+ * @param error The underying error to categorize.
+ */
+export function mapErrorToType(errorMap: ErrorMap, error: Error): ErrorType {
+	const errorTypes = [...Object.keys(errorMap)] as ErrorType[];
+	for (const errorType of errorTypes) {
+		const matcher = errorMap[errorType];
+		if (matcher(error)) {
+			return errorType;
+		}
+	}
+	return 'Unknown';
+}
+
 export enum TransformerMutationType {
 	CREATE = 'Create',
 	UPDATE = 'Update',

packages/datastore/src/types.ts

@@ -658,7 +658,7 @@ export type DataStoreConfig = {
 	DataStore?: {
 		authModeStrategyType?: AuthModeStrategyType;
 		conflictHandler?: ConflictHandler; // default : retry until client wins up to x times
-		errorHandler?: (error: SyncError) => void; // default : logger.warn
+		errorHandler?: (error: SyncError<PersistentModel>) => void; // default : logger.warn
 		maxRecordsToSync?: number; // merge
 		syncPageSize?: number;
 		fullSyncInterval?: number;
@@ -668,7 +668,7 @@ export type DataStoreConfig = {
 	};
 	authModeStrategyType?: AuthModeStrategyType;
 	conflictHandler?: ConflictHandler; // default : retry until client wins up to x times
-	errorHandler?: (error: SyncError) => void; // default : logger.warn
+	errorHandler?: (error: SyncError<PersistentModel>) => void; // default : logger.warn
 	maxRecordsToSync?: number; // merge
 	syncPageSize?: number;
 	fullSyncInterval?: number;
@@ -775,15 +775,32 @@ export type SyncConflict = {
 	attempts: number;
 };
 
-export type SyncError = {
+export type SyncError<T extends PersistentModel> = {
 	message: string;
-	errorType: string;
-	errorInfo: string;
-	localModel: PersistentModel;
-	remoteModel: PersistentModel;
+	errorType: ErrorType;
+	errorInfo?: string;
+	recoverySuggestion?: string;
+	model?: string;
+	localModel: T;
+	remoteModel: T;
+	process: ProcessName;
 	operation: string;
+	cause?: Error;
 };
 
+export type ErrorType =
+	| 'ConfigError'
+	| 'BadRecord'
+	| 'Unauthorized'
+	| 'Transient'
+	| 'Unknown';
+
+export enum ProcessName {
+	'sync' = 'sync',
+	'mutate' = 'mutate',
+	'subscribe' = 'subscribe',
+}
+
 export const DISCARD = Symbol('DISCARD');
 
 export type ConflictHandler = (
@@ -792,7 +809,7 @@ export type ConflictHandler = (
 	| Promise<PersistentModel | typeof DISCARD>
 	| PersistentModel
 	| typeof DISCARD;
-export type ErrorHandler = (error: SyncError) => void;
+export type ErrorHandler = (error: SyncError<PersistentModel>) => void;
 
 export type DeferredCallbackResolverOptions = {
 	callback: () => void;