Refactor RateLimitedScheduler to share pool with advisories download

Author: sigurdm

lib/src/rate_limited_scheduler.dart

@@ -60,11 +60,9 @@ class RateLimitedScheduler<J, V> {
   /// Jobs that have started running.
   final Set<J> _started = {};
 
-  RateLimitedScheduler(
-    Future<V> Function(J) runJob, {
-    required int maxConcurrentOperations,
-  }) : _runJob = runJob,
-       _pool = Pool(maxConcurrentOperations);
+  RateLimitedScheduler(Future<V> Function(J) runJob, {required Pool pool})
+    : _runJob = runJob,
+      _pool = pool;
 
   /// Pick the next task in [_queue] and run it.
   ///

lib/src/source/hosted.dart

@@ -582,23 +582,21 @@ class HostedSource extends CachedSource {
     final Map<String, dynamic> body;
     final List<Advisory>? result;
     try {
-      bodyText = await cache.hostedCache.advisoriesDownloadPool.withResource(
-        () async {
-          return await withAuthenticatedClient(cache, Uri.parse(hostedUrl), (
-            client,
-          ) async {
-            return await retryForHttp(
-              'fetching advisories for "$packageName" from "$url"',
-              () async {
-                final request = http.Request('GET', url);
-                request.attachPubApiHeaders();
-                final response = await client.fetch(request);
-                return response.body;
-              },
-            );
-          });
-        },
-      );
+      bodyText = await cache.hostedCache.pool.withResource(() async {
+        return await withAuthenticatedClient(cache, Uri.parse(hostedUrl), (
+          client,
+        ) async {
+          return await retryForHttp(
+            'fetching advisories for "$packageName" from "$url"',
+            () async {
+              final request = http.Request('GET', url);
+              request.attachPubApiHeaders();
+              final response = await client.fetch(request);
+              return response.body;
+            },
+          );
+        });
+      });
       final decoded = jsonDecode(bodyText);
       if (decoded is! Map<String, dynamic>) {
         throw const FormatException('security advisories must be a mapping');
@@ -2177,6 +2175,9 @@ int? _parseCrc32c(Map<String, String> headers, String fileName) {
 
 /// State that is cached for the HostedSource.
 final class HostedSourceCache {
+  /// A pool to rate-limit concurrent network requests to pub.dev.
+  final Pool pool;
+
   /// The scheduler used to fetch version information for packages.
   ///
   /// This allows rate-limiting requests and speculative pre-fetching of
@@ -2196,12 +2197,11 @@ final class HostedSourceCache {
   /// An in-memory cache to store the futures of fetched security advisories.
   final Map<PackageId, Future<List<Advisory>?>> _advisoriesCache = {};
 
-  /// A pool to rate-limit concurrent security advisory network downloads.
-  final Pool advisoriesDownloadPool = Pool(8);
+  HostedSourceCache(HostedSource hosted) : this._(hosted, Pool(10));
 
-  HostedSourceCache(HostedSource hosted)
+  HostedSourceCache._(HostedSource hosted, this.pool)
     : scheduler = RateLimitedScheduler(
         (HostedRefAndCache refAndCache) => hosted.fetchVersions(refAndCache),
-        maxConcurrentOperations: 10,
+        pool: pool,
       );
 }

test/rate_limited_scheduler_test.dart

@@ -4,6 +4,7 @@
 
 import 'dart:async';
 
+import 'package:pool/pool.dart';
 import 'package:pub/src/rate_limited_scheduler.dart';
 import 'package:test/test.dart';
 
@@ -24,7 +25,7 @@ void main() {
       return i.toUpperCase();
     }
 
-    final scheduler = RateLimitedScheduler(f, maxConcurrentOperations: 2);
+    final scheduler = RateLimitedScheduler(f, pool: Pool(2));
     await scheduler.withPrescheduling((preschedule) async {
       preschedule('a');
       preschedule('b');
@@ -53,7 +54,7 @@ void main() {
         return i.toUpperCase();
       }
 
-      final scheduler = RateLimitedScheduler(f, maxConcurrentOperations: 1);
+      final scheduler = RateLimitedScheduler(f, pool: Pool(1));
 
       await scheduler.withPrescheduling((preschedule1) async {
         await scheduler.withPrescheduling((preschedule2) async {
@@ -88,7 +89,7 @@ void main() {
         return i.toUpperCase();
       }
 
-      final scheduler = RateLimitedScheduler(f, maxConcurrentOperations: 1);
+      final scheduler = RateLimitedScheduler(f, pool: Pool(1));
 
       Future? b;
       await scheduler.withPrescheduling((preschedule) async {
@@ -118,7 +119,7 @@ void main() {
       return i.toUpperCase();
     }
 
-    final scheduler = RateLimitedScheduler(f, maxConcurrentOperations: 2);
+    final scheduler = RateLimitedScheduler(f, pool: Pool(2));
 
     completers['a']!.complete();
     expect(await scheduler.schedule('a'), 'A');
@@ -136,7 +137,7 @@ void main() {
       return i.toUpperCase();
     }
 
-    final scheduler = RateLimitedScheduler(f, maxConcurrentOperations: 1);
+    final scheduler = RateLimitedScheduler(f, pool: Pool(1));
     await scheduler.withPrescheduling((preschedule) async {
       preschedule('a');
       preschedule('b');
@@ -161,7 +162,7 @@ void main() {
       return i.toUpperCase();
     }
 
-    final scheduler = RateLimitedScheduler(f, maxConcurrentOperations: 2);
+    final scheduler = RateLimitedScheduler(f, pool: Pool(2));
 
     await scheduler.withPrescheduling((preschedule) async {
       preschedule('a');
@@ -192,7 +193,7 @@ void main() {
       return Zone.current['zoneValue'] as String?;
     }
 
-    final scheduler = RateLimitedScheduler(f, maxConcurrentOperations: 2);
+    final scheduler = RateLimitedScheduler(f, pool: Pool(2));
     await scheduler.withPrescheduling((preschedule) async {
       runZoned(() {
         preschedule('a');