clean up check_pull_request.dart

auto_submit/lib/requests/check_pull_request.dart

@@ -36,85 +36,107 @@ class CheckPullRequest extends CheckRequest {
     );
   }
 
-  ///TODO refactor this method out into the base class.
   /// Process pull request messages from Pubsub.
   Future<Response> process(
     String pubSubSubscription,
     int pubSubPulls,
     int pubSubBatchSize,
   ) async {
-    final rootCrumb = '$CheckPullRequest(root)';
+    final crumb = '$CheckPullRequest(root)';
 
-    final Set<int> processingLog = <int>{};
-    final List<pub.ReceivedMessage> messageList = await pullMessages(
+    final List<pub.ReceivedMessage> messages = await pullMessages(
       pubSubSubscription,
       pubSubPulls,
       pubSubBatchSize,
     );
-    if (messageList.isEmpty) {
-      log.info('$rootCrumb: No messages are pulled.');
-      return Response.ok('No messages are pulled.');
-    }
-
-    log.info('$rootCrumb: Processing ${messageList.length} messages');
 
-    final List<Future<void>> futures = <Future<void>>[];
+    log.info('$crumb: pulled message batch of size ${messages.length}');
 
-    for (pub.ReceivedMessage message in messageList) {
-      log.info('$rootCrumb: ${message.toJson()}');
-      assert(message.message != null);
-      assert(message.message!.data != null);
-      final String messageData = message.message!.data!;
+    if (messages.isEmpty) {
+      log.info('$crumb: nothing to do, exiting.');
+      return Response.ok('$crumb: nothing to do, exiting.');
+    }
 
-      final Map<String, dynamic> rawBody =
-          json.decode(String.fromCharCodes(base64.decode(messageData))) as Map<String, dynamic>;
-      log.info('$rootCrumb: request raw body = $rawBody');
+    final workItems = await _extractPullRequestFromMessages(pubSubSubscription, messages);
 
-      final pullRequest = PullRequest.fromJson(rawBody);
-      final prCrumb = '$CheckPullRequest(${pullRequest.repo?.fullName}/${pullRequest.number})';
+    // Process pull requests in parallel.
+    final List<Future<void>> futures = <Future<void>>[];
+    for (final workItem in workItems) {
+      futures.add(_processPullRequest(workItem.pullRequest, workItem.ackId));
+    }
+    await Future.wait(futures);
 
-      log.info('$prCrumb: Processing message ackId: ${message.ackId}');
-      log.info('$prCrumb: Processing mesageId: ${message.message!.messageId}');
-      log.info('$prCrumb: Processing PR: ${pullRequest.toJson()}');
+    return Response.ok('Finished processing changes');
+  }
 
-      if (processingLog.contains(pullRequest.number)) {
-        // Ack duplicate.
-        log.info('$prCrumb: Ack the duplicated message : ${message.ackId!}.');
-        await pubsub.acknowledge(
-          pubSubSubscription,
-          message.ackId!,
-        );
+  Future<List<({PullRequest pullRequest, String ackId})>> _extractPullRequestFromMessages(
+    String pubSubSubscription,
+    List<pub.ReceivedMessage> messages,
+  ) async {
+    final crumb = '$CheckPullRequest(root)';
+    final workItems = <int, ({PullRequest pullRequest, String ackId})>{};
 
+    for (pub.ReceivedMessage message in messages) {
+      assert(message.message != null);
+      assert(message.message!.data != null);
+      log.info(
+        '$crumb: processing message: '
+        'id = ${message.message?.messageId}, '
+        'ackId = ${message.ackId}, '
+        'JSON = ${json.encode(message.toJson())}',
+      );
+
+      final messageData = message.message!.data!;
+      final requestBodyJson = String.fromCharCodes(base64.decode(messageData));
+      log.info('$crumb: request JSON = $requestBodyJson');
+
+      final requestBody = json.decode(requestBodyJson) as Map<String, Object?>;
+      final pullRequest = PullRequest.fromJson(requestBody);
+
+      if (workItems.containsKey(pullRequest.number)) {
+        // Duplicate pull request. This can happen, for example, when multiple
+        // labels (say, "autosubmit" and "emergency") are added, each inducing a
+        // pubsub message. Such batches do not need to be processed individually
+        // because PullRequestValidationService will consider the entire state
+        // of the PR and decide to submit it or not based on all the labels set
+        // on it. So the message is deduplicated but still ackowledged so it is
+        // not delivered again.
+        log.info('$crumb: deduplicated pull request #${pullRequest.number}');
+        await pubsub.acknowledge(pubSubSubscription, message.ackId!);
         continue;
       } else {
-        final ApproverService approver = approverProvider(config);
-        log.info('$prCrumb: Checking auto approval of pull request: $rawBody');
-        await approver.autoApproval(pullRequest);
-        processingLog.add(pullRequest.number!);
+        workItems[pullRequest.number!] = (pullRequest: pullRequest, ackId: message.ackId!);
       }
+    }
+
+    return workItems.values.toList();
+  }
 
+  Future<void> _processPullRequest(PullRequest pullRequest, String ackId) async {
+    final crumb = '$CheckPullRequest(${pullRequest.repo?.fullName}/${pullRequest.number})';
+    log.info('$crumb: Processing PR: ${pullRequest.toJson()}');
+
+    try {
+      final approver = approverProvider(config);
+      await approver.autoApproval(pullRequest);
+
+      final validationService = PullRequestValidationService(config);
+      await validationService.processMessage(pullRequest, ackId, pubsub);
+    } catch (error, stackTrace) {
       // Log at severe level but do not rethrow. Because this loop processes a
       // batch of messages, one for each pull request, we don't want one pull
       // request to affect the outcome of processing other pull requests.
       // Because each message is acked individually, the successful ones will
       // be acked, and the failed ones will not, and will be retried by pubsub
       // later according to the retry policy set up in Cocoon.
-      Future<void> onError(Object? error, Object? stackTrace) async {
-        log.severe('''$prCrumb: failed to process message.
+      log.severe('''$crumb: failed to process message.
 
 Pull request: https://github.com/${pullRequest.repo?.fullName}/${pullRequest.number}
 Parsed pull request: ${pullRequest.toJson()}
 
 Error: $error
 $stackTrace
 ''');
-      }
-
-      final PullRequestValidationService validationService = PullRequestValidationService(config);
-      final processFuture = validationService.processMessage(pullRequest, message.ackId!, pubsub).catchError(onError);
-      futures.add(processFuture);
     }
-    await Future.wait(futures);
-    return Response.ok('Finished processing changes');
   }
 }