Send messages in AsyncReceiveCommits from the main thread

Fixes a race condition between sending the "done" progress message and
the rest of ReceiveCommits' output, and avoids potentially blocking in
the worker thread.

Change-Id: I782ae467b594971287b73e246552eb64e07c1be5

gerrit-server/src/main/java/com/google/gerrit/server/git/AsyncReceiveCommits.java

@@ -34,6 +34,7 @@ import org.eclipse.jgit.lib.Config;
 import org.eclipse.jgit.lib.Repository;
 import org.eclipse.jgit.transport.PreReceiveHook;
 import org.eclipse.jgit.transport.ReceiveCommand;
+import org.eclipse.jgit.transport.ReceiveCommand.Result;
 import org.eclipse.jgit.transport.ReceivePack;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -167,14 +168,16 @@ public class AsyncReceiveCommits implements PreReceiveHook {
           timeoutMillis, TimeUnit.MILLISECONDS);
     } catch (ExecutionException e) {
       log.warn("Error in ReceiveCommits", e);
-      rc.getMessageSender().sendError("internal error while processing changes");
+      rc.addError("internal error while processing changes");
       // ReceiveCommits has tried its best to catch errors, so anything at this
       // point is very bad.
       for (final ReceiveCommand c : commands) {
-        if (c.getResult() == ReceiveCommand.Result.NOT_ATTEMPTED) {
-          rc.reject(c, "internal error");
+        if (c.getResult() == Result.NOT_ATTEMPTED) {
+          c.setResult(Result.REJECTED_OTHER_REASON, "internal error");
         }
       }
+    } finally {
+      rc.sendMessages();
     }
   }
 

gerrit-server/src/main/java/com/google/gerrit/server/git/ReceiveCommits.java

@@ -405,11 +405,11 @@ public class ReceiveCommits {
     messages.add(new Message(message, false));
   }
 
-  private void addError(String error) {
+  void addError(String error) {
     messages.add(new Message(error, true));
   }
 
-  private void sendMessages() {
+  void sendMessages() {
     for (Message m : messages) {
       if (m.isError) {
         messageSender.sendError(m.message);
@@ -421,86 +421,82 @@ public class ReceiveCommits {
 
   void processCommands(final Collection<ReceiveCommand> commands,
       final MultiProgressMonitor progress) {
-    try {
-      newProgress = progress.beginSubTask("new", UNKNOWN);
-      replaceProgress = progress.beginSubTask("updated", UNKNOWN);
-      closeProgress = progress.beginSubTask("closed", UNKNOWN);
-      commandProgress = progress.beginSubTask("refs", commands.size());
+    newProgress = progress.beginSubTask("new", UNKNOWN);
+    replaceProgress = progress.beginSubTask("updated", UNKNOWN);
+    closeProgress = progress.beginSubTask("closed", UNKNOWN);
+    commandProgress = progress.beginSubTask("refs", commands.size());
 
-      parseCommands(commands);
-      if (newChange != null
-          && newChange.getResult() == ReceiveCommand.Result.NOT_ATTEMPTED) {
-        createNewChanges();
+    parseCommands(commands);
+    if (newChange != null
+        && newChange.getResult() == ReceiveCommand.Result.NOT_ATTEMPTED) {
+      createNewChanges();
+    }
+    newProgress.end();
+
+    doReplaces();
+    replaceProgress.end();
+
+    for (final ReceiveCommand c : commands) {
+      if (c.getResult() == Result.OK) {
+        switch (c.getType()) {
+          case CREATE:
+            if (isHead(c)) {
+              autoCloseChanges(c);
+            }
+            break;
+
+          case UPDATE: // otherwise known as a fast-forward
+            tagCache.updateFastForward(project.getNameKey(),
+                c.getRefName(),
+                c.getOldId(),
+                c.getNewId());
+            if (isHead(c)) {
+              autoCloseChanges(c);
+            }
+            break;
+
+          case UPDATE_NONFASTFORWARD:
+            if (isHead(c)) {
+              autoCloseChanges(c);
+            }
+            break;
+        }
+
+        if (isConfig(c)) {
+          projectCache.evict(project);
+          ProjectState ps = projectCache.get(project.getNameKey());
+          repoManager.setProjectDescription(project.getNameKey(), //
+              ps.getProject().getDescription());
+        }
+
+        if (!MagicBranch.isMagicBranch(c.getRefName())) {
+          // We only schedule direct refs updates for replication.
+          // Change refs are scheduled when they are created.
+          //
+          replication.scheduleUpdate(project.getNameKey(), c.getRefName());
+          Branch.NameKey destBranch = new Branch.NameKey(project.getNameKey(), c.getRefName());
+          hooks.doRefUpdatedHook(destBranch, c.getOldId(), c.getNewId(), currentUser.getAccount());
+        }
+        commandProgress.update(1);
       }
-      newProgress.end();
+    }
+    closeProgress.end();
+    commandProgress.end();
+    progress.end();
 
-      doReplaces();
-      replaceProgress.end();
-
-      for (final ReceiveCommand c : commands) {
-        if (c.getResult() == Result.OK) {
-          switch (c.getType()) {
-            case CREATE:
-              if (isHead(c)) {
-                autoCloseChanges(c);
-              }
-              break;
-
-            case UPDATE: // otherwise known as a fast-forward
-              tagCache.updateFastForward(project.getNameKey(),
-                  c.getRefName(),
-                  c.getOldId(),
-                  c.getNewId());
-              if (isHead(c)) {
-                autoCloseChanges(c);
-              }
-              break;
-
-            case UPDATE_NONFASTFORWARD:
-              if (isHead(c)) {
-                autoCloseChanges(c);
-              }
-              break;
-          }
-
-          if (isConfig(c)) {
-            projectCache.evict(project);
-            ProjectState ps = projectCache.get(project.getNameKey());
-            repoManager.setProjectDescription(project.getNameKey(), //
-                ps.getProject().getDescription());
-          }
-
-          if (!MagicBranch.isMagicBranch(c.getRefName())) {
-            // We only schedule direct refs updates for replication.
-            // Change refs are scheduled when they are created.
-            //
-            replication.scheduleUpdate(project.getNameKey(), c.getRefName());
-            Branch.NameKey destBranch = new Branch.NameKey(project.getNameKey(), c.getRefName());
-            hooks.doRefUpdatedHook(destBranch, c.getOldId(), c.getNewId(), currentUser.getAccount());
-          }
-          commandProgress.update(1);
+    if (!allNewChanges.isEmpty() && canonicalWebUrl != null) {
+      final String url = canonicalWebUrl;
+      addMessage("");
+      addMessage("New Changes:");
+      for (final Change c : allNewChanges) {
+        if (c.getStatus() == Change.Status.DRAFT) {
+          addMessage("  " + url + c.getChangeId() + " [DRAFT]");
+        }
+        else {
+          addMessage("  " + url + c.getChangeId());
         }
       }
-      closeProgress.end();
-      commandProgress.end();
-      progress.end();
-
-      if (!allNewChanges.isEmpty() && canonicalWebUrl != null) {
-        final String url = canonicalWebUrl;
-        addMessage("");
-        addMessage("New Changes:");
-        for (final Change c : allNewChanges) {
-          if (c.getStatus() == Change.Status.DRAFT) {
-            addMessage("  " + url + c.getChangeId() + " [DRAFT]");
-          }
-          else {
-            addMessage("  " + url + c.getChangeId());
-          }
-        }
-        addMessage("");
-      }
-    } finally {
-      sendMessages();
+      addMessage("");
     }
   }
 
@@ -2060,7 +2056,7 @@ public class ReceiveCommits {
     reject(cmd, "prohibited by Gerrit");
   }
 
-  void reject(final ReceiveCommand cmd, final String why) {
+  private void reject(final ReceiveCommand cmd, final String why) {
     cmd.setResult(ReceiveCommand.Result.REJECTED_OTHER_REASON, why);
     commandProgress.update(1);
   }