feat(datastore): add per-model syncPageSize and maxRecordsToSync configuration (#7310)

Allow syncExpression() to accept an optional third parameter with
per-model sync settings (syncPageSize, maxRecordsToSync) that override
global defaults. Models without per-model config continue to use the
global values. This lets users lower the page size for models with
large records (e.g., exceeding AppSync's 1MB response limit) without
penalizing sync performance for all other models.

Author: danrivett

packages/datastore/__tests__/connectivityHandling.test.ts

@@ -1106,6 +1106,112 @@ describe('DataStore sync engine', () => {
 			}
 		`);
 		});
+
+		describe('per-model sync configuration', () => {
+			/**
+			 * These tests verify that per-model syncPageSize and maxRecordsToSync
+			 * overrides (passed as the 3rd argument to syncExpression) are correctly
+			 * applied to sync queries, while models without overrides fall back to
+			 * the global defaults.
+			 *
+			 * graphqlService.requests accumulates across the entire test lifecycle
+			 * (initial sync from beforeEach, data generation, resync). We snapshot
+			 * the request count before resyncWith() and only inspect requests made
+			 * after that point.
+			 */
+
+			const getSyncRequestsAfter = (
+				tableName: string,
+				afterIndex: number,
+			) =>
+				graphqlService.requests.slice(afterIndex).filter(
+					r =>
+						r.operation === 'query' &&
+						r.type === 'sync' &&
+						r.tableName === tableName,
+				);
+
+			test('per-model syncPageSize overrides global default', async () => {
+				(DataStore as any).amplifyConfig.syncPageSize = 1000;
+				(DataStore as any).amplifyConfig.maxRecordsToSync = 10000;
+
+				const requestsBefore = graphqlService.requests.length;
+				await resyncWith([
+					syncExpression(Post, () => Predicates.ALL, {
+						syncPageSize: 50,
+					}),
+				]);
+
+				const postSyncRequests = getSyncRequestsAfter('Post', requestsBefore);
+				expect(postSyncRequests.length).toBeGreaterThan(0);
+				expect(postSyncRequests[0].variables.limit).toBe(50);
+			});
+
+			test('model without per-model config uses global syncPageSize', async () => {
+				(DataStore as any).amplifyConfig.syncPageSize = 500;
+				(DataStore as any).amplifyConfig.maxRecordsToSync = 10000;
+
+				const requestsBefore = graphqlService.requests.length;
+				await resyncWith([
+					syncExpression(Post, () => Predicates.ALL, {
+						syncPageSize: 50,
+					}),
+				]);
+
+				// Comment has no per-model config, should use global
+				const commentSyncRequests = getSyncRequestsAfter(
+					'Comment',
+					requestsBefore,
+				);
+				expect(commentSyncRequests.length).toBeGreaterThan(0);
+				expect(commentSyncRequests[0].variables.limit).toBe(500);
+			});
+
+			test('multiple models with different per-model syncPageSize', async () => {
+				(DataStore as any).amplifyConfig.syncPageSize = 1000;
+				(DataStore as any).amplifyConfig.maxRecordsToSync = 10000;
+
+				const requestsBefore = graphqlService.requests.length;
+				await resyncWith([
+					syncExpression(Post, () => Predicates.ALL, {
+						syncPageSize: 75,
+					}),
+					syncExpression(Model, () => Predicates.ALL, {
+						syncPageSize: 200,
+					}),
+				]);
+
+				const postSyncRequests = getSyncRequestsAfter('Post', requestsBefore);
+				expect(postSyncRequests.length).toBeGreaterThan(0);
+				expect(postSyncRequests[0].variables.limit).toBe(75);
+
+				const modelSyncRequests = getSyncRequestsAfter(
+					'Model',
+					requestsBefore,
+				);
+				expect(modelSyncRequests.length).toBeGreaterThan(0);
+				expect(modelSyncRequests[0].variables.limit).toBe(200);
+			});
+
+			test('per-model maxRecordsToSync caps limit below syncPageSize', async () => {
+				(DataStore as any).amplifyConfig.syncPageSize = 1000;
+				(DataStore as any).amplifyConfig.maxRecordsToSync = 10000;
+
+				const requestsBefore = graphqlService.requests.length;
+				await resyncWith([
+					syncExpression(Post, () => Predicates.ALL, {
+						syncPageSize: 1000,
+						maxRecordsToSync: 25,
+					}),
+				]);
+
+				const postSyncRequests = getSyncRequestsAfter('Post', requestsBefore);
+				expect(postSyncRequests.length).toBeGreaterThan(0);
+				// limit = Math.min(maxRecordsToSync - 0, syncPageSize) = Math.min(25, 1000) = 25
+				expect(postSyncRequests[0].variables.limit).toBe(25);
+			});
+
+		});
 	});
 
 	describe('error handling', () => {

packages/datastore/__tests__/syncExpression.test.ts

@@ -0,0 +1,56 @@
+import { syncExpression, PerModelSyncConfig } from '../src/types';
+import { Predicates } from '../src/predicates';
+
+// Minimal fake model constructor — syncExpression only stores the reference,
+// it doesn't inspect it.
+const FakeModel: any = function FakeModel() {};
+FakeModel.copyOf = () => {};
+
+describe('syncExpression', () => {
+	test('returns syncConfig when provided', async () => {
+		const config: PerModelSyncConfig = { syncPageSize: 50 };
+		const result = await syncExpression(
+			FakeModel,
+			() => Predicates.ALL,
+			config,
+		);
+
+		expect(result.syncConfig).toEqual({ syncPageSize: 50 });
+		expect(result.modelConstructor).toBe(FakeModel);
+	});
+
+	test('returns syncConfig with both fields', async () => {
+		const config: PerModelSyncConfig = {
+			syncPageSize: 100,
+			maxRecordsToSync: 500,
+		};
+		const result = await syncExpression(
+			FakeModel,
+			() => Predicates.ALL,
+			config,
+		);
+
+		expect(result.syncConfig).toEqual({
+			syncPageSize: 100,
+			maxRecordsToSync: 500,
+		});
+	});
+
+	test('syncConfig is undefined when omitted (backward compatible)', async () => {
+		const result = await syncExpression(FakeModel, () => Predicates.ALL);
+
+		expect(result.syncConfig).toBeUndefined();
+		expect(result.modelConstructor).toBe(FakeModel);
+		expect(result.conditionProducer).toBeDefined();
+	});
+
+	test('syncConfig is undefined when explicitly passed as undefined', async () => {
+		const result = await syncExpression(
+			FakeModel,
+			() => Predicates.ALL,
+			undefined,
+		);
+
+		expect(result.syncConfig).toBeUndefined();
+	});
+});

packages/datastore/src/datastore/datastore.ts

@@ -50,6 +50,7 @@ import {
 	NonModelTypeConstructor,
 	ObserveQueryOptions,
 	PaginationInput,
+	PerModelSyncConfig,
 	PersistentModel,
 	PersistentModelConstructor,
 	PersistentModelMetaData,
@@ -1551,7 +1552,10 @@ class DataStore {
 						aws_appsync_graphqlEndpoint,
 					);
 
-					this.syncPredicates = await this.processSyncExpressions();
+					const { syncPredicates, perModelSyncConfig } =
+						await this.processSyncExpressions();
+					this.syncPredicates = syncPredicates;
+					this.amplifyConfig.perModelSyncConfig = perModelSyncConfig;
 					this.sync = new SyncEngine(
 						schema,
 						namespaceResolver,
@@ -2675,22 +2679,40 @@ class DataStore {
 
 	/**
 	 * Examines the configured `syncExpressions` and produces a WeakMap of
-	 * SchemaModel -> predicate to use during sync.
+	 * SchemaModel -> predicate to use during sync, and a WeakMap of
+	 * SchemaModel -> per-model sync config.
 	 */
-	private async processSyncExpressions(): Promise<
-		WeakMap<SchemaModel, ModelPredicate<any> | null>
-	> {
+	private async processSyncExpressions(): Promise<{
+		syncPredicates: WeakMap<SchemaModel, ModelPredicate<any> | null>;
+		perModelSyncConfig: WeakMap<SchemaModel, PerModelSyncConfig>;
+	}> {
 		if (!this.syncExpressions || !this.syncExpressions.length) {
-			return new WeakMap<SchemaModel, ModelPredicate<any>>();
+			return {
+				syncPredicates: new WeakMap<SchemaModel, ModelPredicate<any>>(),
+				perModelSyncConfig: new WeakMap<SchemaModel, PerModelSyncConfig>(),
+			};
 		}
 
+		const perModelSyncConfig = new WeakMap<SchemaModel, PerModelSyncConfig>();
+
 		const syncPredicates = await Promise.all(
 			this.syncExpressions.map(
 				async (
 					syncExpression: SyncExpression,
 				): Promise<[SchemaModel, ModelPredicate<any> | null]> => {
-					const { modelConstructor, conditionProducer } = await syncExpression;
-					const modelDefinition = getModelDefinition(modelConstructor)!;
+					const { modelConstructor, conditionProducer, syncConfig } =
+						await syncExpression;
+					const modelDefinition = getModelDefinition(modelConstructor);
+
+					if (!modelDefinition) {
+						throw new Error(
+							`Invalid model provided to syncExpression: ${modelConstructor?.name}; unable to find model definition.`,
+						);
+					}
+
+					if (syncConfig) {
+						perModelSyncConfig.set(modelDefinition, syncConfig);
+					}
 
 					// conditionProducer is either a predicate, e.g. (c) => c.field.eq(1)
 					// OR a function/promise that returns a predicate
@@ -2714,7 +2736,10 @@ class DataStore {
 			),
 		);
 
-		return this.weakMapFromEntries(syncPredicates);
+		return {
+			syncPredicates: this.weakMapFromEntries(syncPredicates),
+			perModelSyncConfig,
+		};
 	}
 
 	private async unwrapPromise<T extends PersistentModel>(

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

@@ -363,7 +363,11 @@ class SyncProcessor {
 	start(
 		typesLastSync: Map<SchemaModel, [string, number]>,
 	): Observable<SyncModelPage> {
-		const { maxRecordsToSync, syncPageSize } = this.amplifyConfig;
+		const {
+			maxRecordsToSync: defaultMaxRecordsToSync,
+			syncPageSize: defaultSyncPageSize,
+			perModelSyncConfig,
+		} = this.amplifyConfig;
 		const parentPromises = new Map<string, Promise<void>>();
 		const observable = new Observable<SyncModelPage>(observer => {
 			const sortedTypesLastSyncs = Object.values(this.schema.namespaces).reduce(
@@ -394,6 +398,12 @@ class SyncProcessor {
 							let recordsReceived = 0;
 							const filter = this.graphqlFilterFromPredicate(modelDefinition);
 
+							const modelSyncConfig = perModelSyncConfig?.get(modelDefinition);
+							const syncPageSize =
+								modelSyncConfig?.syncPageSize ?? defaultSyncPageSize;
+							const maxRecordsToSync =
+								modelSyncConfig?.maxRecordsToSync ?? defaultMaxRecordsToSync;
+
 							const parents = this.schema.namespaces[
 								namespace
 							].modelTopologicalOrdering!.get(modelDefinition.name);

packages/datastore/src/types.ts

@@ -1093,9 +1093,15 @@ export type ModelAuthModes = Record<
 	Record<ModelOperation, GraphQLAuthMode[]>
 >;
 
+export interface PerModelSyncConfig {
+	syncPageSize?: number;
+	maxRecordsToSync?: number;
+}
+
 export type SyncExpression = Promise<{
 	modelConstructor: any;
 	conditionProducer(c?: any): any;
+	syncConfig?: PerModelSyncConfig;
 }>;
 
 /*
@@ -1156,6 +1162,7 @@ type ConditionProducer<T extends PersistentModel, A extends Option<T>> = (
  *
  * @param modelConstructor The Model from the schema.
  * @param conditionProducer A function that builds a condition object that can describe how to filter the model.
+ * @param syncConfig Optional per-model sync configuration (e.g., `syncPageSize`, `maxRecordsToSync`).
  * @returns An sync expression object that can be attached to the DataStore `syncExpressions` configuration property.
  */
 export async function syncExpression<
@@ -1164,13 +1171,16 @@ export async function syncExpression<
 >(
 	modelConstructor: PersistentModelConstructor<T>,
 	conditionProducer: ConditionProducer<T, A>,
+	syncConfig?: PerModelSyncConfig,
 ): Promise<{
 	modelConstructor: PersistentModelConstructor<T>;
 	conditionProducer: ConditionProducer<T, A>;
+	syncConfig?: PerModelSyncConfig;
 }> {
 	return {
 		modelConstructor,
 		conditionProducer,
+		syncConfig,
 	};
 }