fix(cat-voices): pool concurrent documents sync (#2334)

* fix: pool concurrent documents sync

* chore: update docs

---------

Co-authored-by: Dominik Toton <[email protected]>

Author: damian-molinski

catalyst_voices/melos.yaml

@@ -148,6 +148,7 @@ command:
       convert: ^3.1.1
       path: ^1.9.1
       pinenacl: ^0.6.0
+      pool: ^1.5.1
       uuid_plus: ^0.1.0
       sentry_flutter: ^8.8.0
       scrollable_positioned_list: ^0.3.8

catalyst_voices/packages/internal/catalyst_voices_repositories/lib/src/document/source/document_data_remote_source.dart

@@ -5,8 +5,12 @@ import 'package:catalyst_voices_repositories/src/common/response_mapper.dart';
 import 'package:catalyst_voices_repositories/src/document/document_data_factory.dart';
 import 'package:catalyst_voices_repositories/src/dto/api/document_index_list_dto.dart';
 import 'package:catalyst_voices_repositories/src/dto/api/document_index_query_filters_dto.dart';
+import 'package:flutter/foundation.dart';
 
 final class CatGatewayDocumentDataSource implements DocumentDataRemoteSource {
+  @visibleForTesting
+  static const indexPageSize = 200;
+
   final ApiServices _api;
   final SignedDocumentManager _signedDocumentManager;
 
@@ -61,7 +65,7 @@ final class CatGatewayDocumentDataSource implements DocumentDataRemoteSource {
     final allRefs = <SignedDocumentRef>{};
 
     var page = 0;
-    const maxPerPage = 100;
+    const maxPerPage = indexPageSize;
     var remaining = 0;
 
     do {

catalyst_voices/packages/internal/catalyst_voices_repositories/test/src/document/source/document_data_remote_source_test.dart

@@ -18,6 +18,8 @@ void main() {
   late final ApiServices apiServices;
   late final CatGatewayDocumentDataSource source;
 
+  const maxPageSize = CatGatewayDocumentDataSource.indexPageSize;
+
   setUpAll(() {
     apiServices = ApiServices.internal(
       gateway: gateway,
@@ -43,17 +45,17 @@ void main() {
         // Given
         final pageZero = DocumentIndexList(
           docs: List.generate(
-            100,
+            maxPageSize,
             (_) => _buildDocumentIndexList().toJson(),
           ),
-          page: const CurrentPage(page: 0, limit: 100, remaining: 5),
+          page: const CurrentPage(page: 0, limit: maxPageSize, remaining: 5),
         );
         final pageOne = DocumentIndexList(
           docs: List.generate(
             5,
             (_) => _buildDocumentIndexList().toJson(),
           ),
-          page: const CurrentPage(page: 1, limit: 100, remaining: 0),
+          page: const CurrentPage(page: 1, limit: maxPageSize, remaining: 0),
         );
 
         final pageZeroResponse = Response(http.Response('', 200), pageZero);
@@ -63,14 +65,14 @@ void main() {
         when(
           () => gateway.apiV1DocumentIndexPost(
             body: const DocumentIndexQueryFilter(),
-            limit: 100,
+            limit: maxPageSize,
             page: 0,
           ),
         ).thenAnswer((_) => Future.value(pageZeroResponse));
         when(
           () => gateway.apiV1DocumentIndexPost(
             body: const DocumentIndexQueryFilter(),
-            limit: 100,
+            limit: maxPageSize,
             page: 1,
           ),
         ).thenAnswer((_) => Future.value(pageOneResponse));
@@ -111,7 +113,7 @@ void main() {
               }).toList(),
             ).toJson(),
           ],
-          page: const CurrentPage(page: 0, limit: 100, remaining: 0),
+          page: const CurrentPage(page: 0, limit: maxPageSize, remaining: 0),
         );
         final response = Response(http.Response('', 200), page);
 
@@ -124,7 +126,7 @@ void main() {
         when(
           () => gateway.apiV1DocumentIndexPost(
             body: const DocumentIndexQueryFilter(),
-            limit: 100,
+            limit: maxPageSize,
             page: 0,
           ),
         ).thenAnswer((_) => Future.value(response));

catalyst_voices/packages/internal/catalyst_voices_services/lib/src/documents/documents_service.dart

@@ -4,6 +4,7 @@ import 'package:catalyst_voices_models/catalyst_voices_models.dart';
 import 'package:catalyst_voices_repositories/catalyst_voices_repositories.dart';
 import 'package:catalyst_voices_shared/catalyst_voices_shared.dart';
 import 'package:flutter/foundation.dart';
+import 'package:pool/pool.dart';
 import 'package:result_type/result_type.dart';
 
 final _logger = Logger('DocumentsService');
@@ -25,6 +26,7 @@ abstract interface class DocumentsService {
   /// Returns list of added refs.
   Future<List<SignedDocumentRef>> sync({
     ValueChanged<double>? onProgress,
+    int maxConcurrent,
   });
 }
 
@@ -38,6 +40,7 @@ final class DocumentsServiceImpl implements DocumentsService {
   @override
   Future<List<SignedDocumentRef>> sync({
     ValueChanged<double>? onProgress,
+    int maxConcurrent = 100,
   }) async {
     onProgress?.call(0.1);
 
@@ -60,31 +63,43 @@ final class DocumentsServiceImpl implements DocumentsService {
 
     var completed = 0;
     final total = missingRefs.length;
-
-    // Note. Handling or errors as Outcome because we have to
-    // give a change to all refs to finish and keep all info about what
-    // failed.
-    final futures = <Future<Result<SignedDocumentRef, Exception>>>[
-      for (final ref in missingRefs)
-        _documentRepository
-            .cacheDocument(ref: ref)
-            .then<Result<SignedDocumentRef, Exception>>((_) => Success(ref))
-            .catchError((Object error, StackTrace stackTrace) {
+    final pool = Pool(maxConcurrent);
+
+    final futures = <Future<void>>[];
+    final outcomes = <Result<SignedDocumentRef, Exception>>[];
+
+    /// Handling or errors as Outcome because we have to
+    /// give a change to all refs to finish and keep all info about what
+    /// failed.
+    for (final ref in missingRefs) {
+      /// Its possible that missingRefs can be very large
+      /// and executing too many requests at once throws
+      /// net::ERR_INSUFFICIENT_RESOURCES in chrome.
+      /// That's reason for adding pool and limiting max requests.
+      final future = pool.withResource<void>(() async {
+        try {
+          await _documentRepository.cacheDocument(ref: ref);
+          outcomes.add(_RefSuccess(ref));
+        } catch (error, stackTrace) {
           final exception = RefSyncException(
             ref,
             error: error,
             stack: stackTrace,
           );
-          return _RefFailure(exception);
-        }).whenComplete(() {
+          outcomes.add(_RefFailure(exception));
+        } finally {
           completed += 1;
           final progress = completed / total;
           final totalProgress = 0.2 + (progress * 0.8);
           onProgress?.call(totalProgress);
-        }),
-    ];
+        }
+      });
+
+      futures.add(future);
+    }
 
-    final outcomes = await futures.wait;
+    // Wait for all operations managed by the pool to complete
+    await Future.wait(futures);
     final failures = outcomes.whereType<_RefFailure>();
 
     if (failures.isNotEmpty) {

catalyst_voices/packages/internal/catalyst_voices_services/pubspec.yaml

@@ -29,6 +29,7 @@ dependencies:
     sdk: flutter
   logging: ^1.3.0
   path: ^1.9.1
+  pool: ^1.5.1
   result_type: ^0.2.0
   rxdart: ^0.27.7
   shared_preferences: ^2.3.3