Revert "Revert "fix: decrease error handler verbosity on self recover…

…ing errors (#9987)" (#10004)"

This reverts commit eb73ad7.

packages/datastore/__tests__/helpers.ts

@@ -994,3 +994,13 @@ export function internalTestSchema(): InternalSchema {
 		version: '1',
 	};
 }
+
+export function smallTestSchema(): Schema {
+	const schema = testSchema();
+	return {
+		...schema,
+		models: {
+			Model: schema.models.Model,
+		},
+	};
+}

packages/datastore/__tests__/subscription.test.ts

@@ -1,9 +1,31 @@
+import Amplify from 'aws-amplify';
 import { GRAPHQL_AUTH_MODE } from '@aws-amplify/api';
+import { CONTROL_MSG as PUBSUB_CONTROL_MSG } from '@aws-amplify/pubsub';
+import Observable from 'zen-observable-ts';
 import {
 	SubscriptionProcessor,
 	USER_CREDENTIALS,
 } from '../src/sync/processors/subscription';
-import { SchemaModel } from '../src/types';
+import {
+	internalTestSchema,
+	Model as ModelType,
+	smallTestSchema,
+} from './helpers';
+import {
+	SchemaModel,
+	InternalSchema,
+	PersistentModelConstructor,
+} from '../src/types';
+
+let mockObservable = new Observable(() => {});
+
+// mock graphql to return a mockable observable
+jest.mock('@aws-amplify/api', () => {
+	return {
+		...jest.requireActual('@aws-amplify/api'),
+		graphql: jest.fn(() => mockObservable),
+	};
+});
 
 describe('sync engine subscription module', () => {
 	test('owner authorization', () => {
@@ -566,6 +588,117 @@ describe('sync engine subscription module', () => {
 	});
 });
 
+describe('error handler', () => {
+	let Model: PersistentModelConstructor<ModelType>;
+
+	let subscriptionProcessor: SubscriptionProcessor;
+	const errorHandler = jest.fn();
+	beforeEach(async () => {
+		errorHandler.mockClear();
+		subscriptionProcessor = await instantiateSubscriptionProcessor({
+			errorHandler,
+		});
+	});
+
+	test('error handler once after all retires have failed', done => {
+		Amplify.Logger.LOG_LEVEL = 'DEBUG';
+		const debugLog = jest.spyOn(console, 'log');
+		const message = PUBSUB_CONTROL_MSG.REALTIME_SUBSCRIPTION_INIT_ERROR;
+		mockObservable = new Observable(observer => {
+			observer.error({
+				error: {
+					errors: [
+						{
+							message,
+						},
+					],
+				},
+			});
+		});
+
+		const subscription = subscriptionProcessor.start();
+		subscription[0].subscribe({
+			error: data => {
+				console.log(data);
+				console.log(errorHandler.mock.calls);
+
+				// call once each for Create, Update, and Delete
+				expect(errorHandler).toHaveBeenCalledTimes(3);
+				['Create', 'Update', 'Delete'].forEach(operation => {
+					expect(errorHandler).toHaveBeenCalledWith(
+						expect.objectContaining({
+							process: 'subscribe',
+							errorType: 'Unknown',
+							message,
+							model: 'Model',
+							operation,
+						})
+					);
+					// expect logger.debug to be called 6 times for auth mode (2 for each operation)
+					// can't use toHaveBeenCalledTimes because it is called elsewhere unrelated to the test
+					expect(debugLog).toHaveBeenCalledWith(
+						expect.stringMatching(
+							new RegExp(
+								`[DEBUG].*${operation} subscription failed with authMode: API_KEY`
+							)
+						)
+					);
+					expect(debugLog).toHaveBeenCalledWith(
+						expect.stringMatching(
+							new RegExp(
+								`[DEBUG].*${operation} subscription failed with authMode: AMAZON_COGNITO_USER_POOLS`
+							)
+						)
+					);
+				});
+
+				done();
+			},
+		});
+	}, 500);
+
+	async function instantiateSubscriptionProcessor({
+		errorHandler = () => null,
+	}) {
+		let schema: InternalSchema = internalTestSchema();
+
+		const { initSchema, DataStore } = require('../src/datastore/datastore');
+		const classes = initSchema(smallTestSchema());
+
+		({ Model } = classes as {
+			Model: PersistentModelConstructor<ModelType>;
+		});
+
+		const userClasses = {
+			Model,
+		};
+
+		await DataStore.start();
+		({ schema } = (DataStore as any).storage.storage);
+		const syncPredicates = new WeakMap();
+
+		const subscriptionProcessor = new SubscriptionProcessor(
+			schema,
+			syncPredicates,
+			{
+				aws_project_region: 'us-west-2',
+				aws_appsync_graphqlEndpoint:
+					'https://xxxxxxxxxxxxxxxxxxxxxx.appsync-api.us-west-2.amazonaws.com/graphql',
+				aws_appsync_region: 'us-west-2',
+				aws_appsync_authenticationType: 'API_KEY',
+				aws_appsync_apiKey: 'da2-xxxxxxxxxxxxxxxxxxxxxx',
+			},
+			() => [
+				GRAPHQL_AUTH_MODE.API_KEY,
+				GRAPHQL_AUTH_MODE.AMAZON_COGNITO_USER_POOLS,
+			],
+			errorHandler
+		);
+
+		return subscriptionProcessor;
+	}
+});
+
 const accessTokenPayload = {
 	sub: 'xxxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxx',
 	'cognito:groups': ['mygroup'],

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

@@ -394,7 +394,9 @@ class SubscriptionProcessor {
 									transformerMutationType
 								].push(
 									queryObservable
-										.map(({ value }) => value)
+										.map(({ value }) => {
+											return value;
+										})
 										.subscribe({
 											next: ({ data, errors }) => {
 												if (Array.isArray(errors) && errors.length > 0) {
@@ -446,23 +448,6 @@ class SubscriptionProcessor {
 														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,
-														model: modelDefinition.name,
-														operation,
-														errorType:
-															getSubscriptionErrorType(subscriptionError),
-														process: ProcessName.subscribe,
-														remoteModel: null,
-														cause: subscriptionError,
-													});
-												} catch (e) {
-													logger.error('Sync error handler failed with:', e);
-												}
 
 												if (
 													message.includes(
@@ -483,16 +468,17 @@ class SubscriptionProcessor {
 														operationAuthModeAttempts[operation] >=
 														readAuthModes.length
 													) {
+														// last auth mode retry. Continue with error
 														logger.debug(
 															`${operation} subscription failed with authMode: ${
 																readAuthModes[
 																	operationAuthModeAttempts[operation] - 1
 																]
 															}`
 														);
-														logger.warn('subscriptionError', message);
-														return;
 													} else {
+														// retry with different auth mode. Do not trigger
+														// observer error or error handler
 														logger.debug(
 															`${operation} subscription failed with authMode: ${
 																readAuthModes[
@@ -508,7 +494,27 @@ class SubscriptionProcessor {
 														return;
 													}
 												}
-												logger.warn('subscriptionError', message);
+
+												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,
+														model: modelDefinition.name,
+														operation,
+														errorType:
+															getSubscriptionErrorType(subscriptionError),
+														process: ProcessName.subscribe,
+														remoteModel: null,
+														cause: subscriptionError,
+													});
+												} catch (e) {
+													logger.error(
+														'Subscription error handler failed with:',
+														e
+													);
+												}
 
 												if (typeof subscriptionReadyCallback === 'function') {
 													subscriptionReadyCallback();